“

Series: S12 Developer Tutorials · Article 12 | Level: Beginner | Reading Time: 15 min

Data Import Guide

#TL;DR

Data import is the first step in using the coomia-dip platform. This guide covers how to import data from relational databases, CSV/Excel files, APIs, message queues, and object storage into the platform, transforming them into Ontology objects. It covers connector configuration, schema mapping, data validation, and common troubleshooting.

#1. Data Import Overview

#1.1 Supported Data Sources

Category	Data Sources	Connection Method
Relational DB	MySQL, PostgreSQL, Oracle, SQL Server	JDBC
Files	CSV, Excel, JSON, Parquet	File Upload / S3
API	REST API, GraphQL	HTTP Connector
Message Queue	Kafka, RabbitMQ, Pulsar	Stream Connector
Object Storage	MinIO, S3, OSS	S3 Protocol
NoSQL	MongoDB, Redis, Elasticsearch	Native Driver

#1.2 Import Workflow

Code

① Register Data Source Connection
    ↓
② Create Schema Mapping
    ↓
③ Configure Data Validation
    ↓
④ Execute Data Sync
    ↓
⑤ Verify Ontology Objects

#2. Relational Database Import

#2.1 Register Database Connection

Python

from ontology_sdk import OntoPlatform
from ontology_sdk.connectors import JdbcConnector

platform = OntoPlatform(
    control_plane_url="localhost:50051",
    data_plane_url="localhost:50052"
)

# Register MySQL connection
mysql_conn = platform.connectors.register(
    JdbcConnector(
        name="hr_mysql",
        display_name="HR System MySQL",
        driver="mysql",
        host="hr-db.internal",
        port=3306,
        database="hr_system",
        username="${HR_DB_USER}",
        password="${HR_DB_PASS}",
        connection_pool={
            "min_size": 5,
            "max_size": 20,
            "timeout": 30,
        }
    )
)

# Test connection
test = platform.connectors.test("hr_mysql")
print(f"Connection test: {'Success' if test.success else 'Failed'}")
print(f"Latency: {test.latency_ms}ms")
print(f"Version: {test.server_version}")

#2.2 Schema Discovery

Python

# Auto-discover database schema
discovery = platform.connectors.discover("hr_mysql")

print("=== Discovered Tables ===")
for table in discovery.tables:
    print(f"\nTable: {table.name} ({table.row_count} rows)")
    for col in table.columns:
        print(f"  {col.name}: {col.type} "
              f"{'NOT NULL' if col.not_null else 'NULLABLE'} "
              f"{'PK' if col.is_primary_key else ''}")

# Suggested Object Type mappings
suggestions = platform.connectors.suggest_mapping("hr_mysql")
for s in suggestions:
    print(f"\nSuggestion: {s.table_name} -> {s.suggested_object_type}")
    for prop in s.property_mappings:
        print(f"  {prop.column} ({prop.column_type}) -> "
              f"{prop.suggested_property} ({prop.suggested_type})")

#2.3 Configure Mapping and Import

Python

from ontology_sdk.connectors import ImportConfig, FieldMapping

config = ImportConfig(
    connector="hr_mysql",
    source_table="employees",
    target_object_type="Employee",
    upsert_key="employee_id",
    mappings=[
        FieldMapping("employee_id", "external_id"),
        FieldMapping("full_name", "name"),
        FieldMapping("email_address", "email"),
        FieldMapping("dept_code", "department",
                     lookup={"object_type": "Department", "match_field": "code", "return": "rid"}),
        FieldMapping("base_salary", "salary", transform="round(value, 2)"),
        FieldMapping("hire_date", "hire_date", transform="parse_date(value, 'yyyy-MM-dd')"),
        FieldMapping("job_level", "level",
                     transform={
                         "J1": "junior", "J2": "junior",
                         "M1": "mid", "M2": "mid",
                         "S1": "senior", "S2": "senior",
                     }),
    ],
    filter="status = 'active'",
    batch_size=500,
)

result = platform.connectors.import_data(config)
print(f"Import complete: {result.total_rows} rows")
print(f"  Succeeded: {result.succeeded}")
print(f"  Skipped: {result.skipped}")
print(f"  Failed: {result.failed}")
print(f"  Duration: {result.duration_seconds}s")

#3. File Import

#3.1 CSV File Import

Python

from ontology_sdk.connectors import FileImporter

importer = FileImporter(platform)

# Single file import
result = importer.import_csv(
    file_path="/data/customers.csv",
    target_object_type="Customer",
    upsert_key="customer_id",
    encoding="utf-8",
    delimiter=",",
    header_row=1,
    mappings={
        "customer_id": "external_id",
        "company_name": "name",
        "industry": "industry",
        "country": "region",
        "annual_revenue": ("annual_revenue", float),
    },
    skip_empty_rows=True,
    on_error="skip",
)

print(f"CSV import: {result.succeeded}/{result.total_rows} succeeded")

# Preview (no actual import)
preview = importer.preview_csv("/data/customers.csv", limit=5)
for row in preview.rows:
    print(row)
print(f"Total rows: {preview.total_rows}")
print(f"Columns: {preview.columns}")

#3.2 Excel File Import

Python

result = importer.import_excel(
    file_path="/data/products.xlsx",
    sheet_name="Sheet1",
    target_object_type="Product",
    upsert_key="sku",
    header_row=1,
    data_start_row=2,
    mappings={
        "A": ("sku", str),
        "B": ("name", str),
        "C": ("category", str),
        "D": ("price", float),
        "E": ("stock_quantity", int),
    },
)

#3.3 JSON File Import

Python

result = importer.import_json(
    file_path="/data/orders.json",
    target_object_type="Order",
    upsert_key="order_id",
    json_path="$.data.orders[*]",
    mappings={
        "id": "order_id",
        "customer.id": "customer_external_id",
        "total": ("total_amount", float),
        "status": "status",
        "created_at": ("created_at", "parse_datetime"),
    },
)

#3.4 Batch File Import

Python

# Batch import CSVs from directory
result = importer.import_directory(
    directory="/data/daily_exports/",
    file_pattern="employees_*.csv",
    target_object_type="Employee",
    upsert_key="employee_id",
    mappings=standard_employee_mapping,
    parallel=4,
)

print(f"Processed {result.file_count} files")
print(f"Total records: {result.total_rows}")
print(f"Succeeded: {result.succeeded}")

#4. API Data Import

#4.1 REST API Connector

Python

from ontology_sdk.connectors import ApiConnector, ApiImportConfig

api_conn = platform.connectors.register(
    ApiConnector(
        name="crm_api",
        display_name="CRM API",
        base_url="https://crm-api.internal/v2",
        auth={
            "type": "bearer",
            "token": "${CRM_API_TOKEN}",
        },
        headers={
            "Content-Type": "application/json",
            "X-API-Version": "2.0",
        },
        rate_limit={
            "requests_per_second": 10,
            "burst": 20,
        },
        timeout_seconds=30,
        retry={
            "max_retries": 3,
            "backoff": "exponential",
        },
    )
)

config = ApiImportConfig(
    connector="crm_api",
    endpoint="/customers",
    method="GET",
    pagination={
        "type": "cursor",
        "cursor_param": "cursor",
        "cursor_path": "$.meta.next_cursor",
        "data_path": "$.data",
        "page_size": 100,
    },
    target_object_type="Customer",
    upsert_key="external_id",
    mappings={
        "id": "external_id",
        "attributes.name": "name",
        "attributes.industry": "industry",
        "attributes.region": "region",
        "attributes.revenue": ("annual_revenue", float),
        "attributes.tier": "tier",
    },
)

result = platform.connectors.import_data(config)
print(f"API import: {result.succeeded} records")

#4.2 Pagination Strategies

Python

# Offset pagination
pagination_offset = {
    "type": "offset",
    "limit_param": "limit",
    "offset_param": "offset",
    "data_path": "$.results",
    "total_path": "$.total",
    "page_size": 100,
}

# Cursor pagination
pagination_cursor = {
    "type": "cursor",
    "cursor_param": "after",
    "cursor_path": "$.pagination.next_cursor",
    "has_more_path": "$.pagination.has_more",
    "data_path": "$.data",
}

# Page number pagination
pagination_page = {
    "type": "page_number",
    "page_param": "page",
    "size_param": "per_page",
    "data_path": "$.items",
    "total_pages_path": "$.total_pages",
    "page_size": 50,
}

#5. Message Queue Import

#5.1 Kafka Real-Time Import

Python

from ontology_sdk.connectors import KafkaConnector

kafka_conn = platform.connectors.register(
    KafkaConnector(
        name="order_kafka",
        display_name="Order Events Kafka",
        bootstrap_servers="kafka-1:9092,kafka-2:9092",
        topics=["order-created", "order-updated", "order-cancelled"],
        group_id="coomia-dip-order-consumer",
        auto_offset_reset="latest",
        security={
            "protocol": "SASL_SSL",
            "mechanism": "PLAIN",
            "username": "${KAFKA_USER}",
            "password": "${KAFKA_PASS}",
        },
    )
)

platform.connectors.start_stream(
    connector="order_kafka",
    target_object_type="Order",
    upsert_key="order_id",
    event_mapping={
        "order-created": {
            "action": "create",
            "mappings": {
                "payload.id": "order_id",
                "payload.customer_id": "customer_external_id",
                "payload.total": ("total_amount", float),
                "payload.status": "status",
            }
        },
        "order-updated": {
            "action": "update",
            "mappings": {
                "payload.id": "order_id",
                "payload.status": "status",
                "payload.updated_at": "updated_at",
            }
        },
        "order-cancelled": {
            "action": "update",
            "key_field": "payload.id",
            "mappings": {
                "payload.id": "order_id",
                "status": {"value": "cancelled"},
                "payload.cancelled_at": "cancelled_at",
            }
        },
    },
    error_handling={
        "dead_letter_topic": "coomia-dip-dlq",
        "max_retries": 3,
    },
)

#6. Object Storage Import

#6.1 MinIO / S3 Import

Python

from ontology_sdk.connectors import S3Connector

s3_conn = platform.connectors.register(
    S3Connector(
        name="data_lake_minio",
        display_name="Data Lake MinIO",
        endpoint="http://minio:9000",
        access_key="${MINIO_ACCESS_KEY}",
        secret_key="${MINIO_SECRET_KEY}",
        region="us-east-1",
        bucket="data-lake",
    )
)

# Import Parquet files
result = platform.connectors.import_s3(
    connector="data_lake_minio",
    path="bronze/customers/2025/03/*.parquet",
    target_object_type="Customer",
    upsert_key="customer_id",
    format="parquet",
    mappings={
        "customer_id": "external_id",
        "name": "name",
        "industry": "industry",
    },
)

# Watch for new files and auto-import
platform.connectors.watch_s3(
    connector="data_lake_minio",
    path_prefix="bronze/daily/",
    file_pattern="*.csv",
    target_object_type="DailyReport",
    upsert_key="report_id",
    poll_interval=60,
    mappings=daily_report_mapping,
)

#7. Data Validation

#7.1 Validation Rules

Python

from ontology_sdk.connectors import ValidationRule

validation = [
    ValidationRule.not_empty("name", on_fail="reject"),
    ValidationRule.email_format("email", on_fail="reject"),
    ValidationRule.range("salary", min=0, max=10000000, on_fail="flag"),
    ValidationRule.regex("phone", r"^\+?[1-9]\d{1,14}$", on_fail="flag"),
    ValidationRule.enum("status", ["active", "inactive", "suspended"], on_fail="reject"),
    ValidationRule.unique("external_id", on_fail="reject"),
    ValidationRule.date_format("hire_date", format="yyyy-MM-dd", on_fail="reject"),
    ValidationRule.not_future("hire_date", on_fail="flag"),
    ValidationRule.custom(
        "age_check",
        lambda row: 18 <= calculate_age(row["birth_date"]) <= 100,
        on_fail="flag",
        message="Age not in reasonable range"
    ),
]

config.validation_rules = validation
config.validation_threshold = 0.95  # Abort if pass rate below 95%

#7.2 Validation Reports

Python

report = platform.connectors.validate(
    connector="hr_mysql",
    source_table="employees",
    rules=validation,
    sample_size=1000,
)

print(f"=== Data Quality Report ===")
print(f"Sample size: {report.sample_size}")
print(f"Pass rate: {report.pass_rate:.1%}")
print()
for rule_result in report.results:
    status = "PASS" if rule_result.pass_rate >= 0.99 else "WARN" if rule_result.pass_rate >= 0.95 else "FAIL"
    print(f"  [{status}] {rule_result.rule_name}: {rule_result.pass_rate:.1%}")
    if rule_result.failures:
        for f in rule_result.failures[:3]:
            print(f"    Row {f.row}: {f.value} - {f.reason}")

#8. Connector Management

Python

# List all connectors
connectors = platform.connectors.list()
for conn in connectors:
    print(f"{conn.name} [{conn.type}] - {conn.display_name}")
    print(f"  Status: {conn.status}")
    print(f"  Last sync: {conn.last_sync_at}")

# Health check
health = platform.connectors.health_check_all()
for name, status in health.items():
    icon = "OK" if status.healthy else "ERR"
    print(f"  [{icon}] {name}: {status.message} ({status.latency_ms}ms)")

# View sync history
history = platform.connectors.get_sync_history("hr_mysql", limit=10)
for run in history:
    print(f"  {run.started_at} | {run.status} | {run.records_synced} records | {run.duration}s")

#9. Common Troubleshooting

Problem	Possible Cause	Solution
Connection timeout	Network/firewall	Check connectivity, open ports
Auth failure	Wrong credentials	Verify username/password, check permissions
Encoding errors	Encoding mismatch	Specify correct encoding (UTF-8/GBK)
Type conversion failure	Incompatible types	Add transform function
Unique key conflict	Duplicate data	Check upsert_key configuration
Missing foreign keys	Dependencies not imported	Import dependent Object Types first
Slow performance	Large data/no index	Increase batch_size, add source indexes

Python

# Enable debug logging
platform.connectors.set_log_level("hr_mysql", "DEBUG")

# Dry run (no writes)
dry_run = platform.connectors.import_data(config, dry_run=True, limit=10)
print(f"Dry run results:")
for sample in dry_run.sample_output:
    print(f"  {sample}")
print(f"Estimated import: {dry_run.estimated_total} records")

#10. Complete Example: Multi-Source Data Import

Python

from ontology_sdk import OntoPlatform
from ontology_sdk.connectors import JdbcConnector, ApiConnector, FileImporter, ImportConfig, FieldMapping

platform = OntoPlatform(
    control_plane_url="localhost:50051",
    data_plane_url="localhost:50052"
)

# === Step 1: Create Object Types ===
platform.schema.create_object_type("Customer", properties={
    "external_id": {"type": "string", "required": True, "indexed": True},
    "name": {"type": "string", "required": True},
    "industry": {"type": "string"},
    "region": {"type": "string"},
    "annual_revenue": {"type": "decimal"},
    "tier": {"type": "string"},
    "source": {"type": "string"},
})

# === Step 2: MySQL - import base customer data ===
platform.connectors.register(JdbcConnector(
    name="crm_mysql", driver="mysql",
    host="crm-db", port=3306, database="crm",
    username="${CRM_USER}", password="${CRM_PASS}",
))

mysql_result = platform.connectors.import_data(ImportConfig(
    connector="crm_mysql",
    source_table="customers",
    target_object_type="Customer",
    upsert_key="external_id",
    mappings=[
        FieldMapping("id", "external_id"),
        FieldMapping("company_name", "name"),
        FieldMapping("industry_code", "industry"),
        FieldMapping("region", "region"),
    ],
    extra_properties={"source": "crm_mysql"},
))
print(f"MySQL import: {mysql_result.succeeded} records")

# === Step 3: API - enrich with revenue data ===
platform.connectors.register(ApiConnector(
    name="billing_api", base_url="https://billing.internal/api",
    auth={"type": "bearer", "token": "${BILLING_TOKEN}"},
))

api_result = platform.connectors.import_data(ImportConfig(
    connector="billing_api",
    endpoint="/customers/revenue",
    target_object_type="Customer",
    upsert_key="external_id",
    mappings=[
        FieldMapping("customer_id", "external_id"),
        FieldMapping("total_revenue", "annual_revenue", transform="float"),
    ],
    merge_mode="update_only",
))
print(f"API enrichment: {api_result.succeeded} records")

# === Step 4: CSV - enrich with tier data ===
importer = FileImporter(platform)
csv_result = importer.import_csv(
    file_path="/data/customer_tiers.csv",
    target_object_type="Customer",
    upsert_key="external_id",
    mappings={
        "customer_id": "external_id",
        "tier": "tier",
    },
    merge_mode="update_only",
)
print(f"CSV enrichment: {csv_result.succeeded} records")

# === Step 5: Verify ===
total = platform.oql.execute("FIND Customer AGGREGATE COUNT(*) AS total")
print(f"\nFinal customer count: {total[0].total}")

quality = platform.oql.execute("""
    FIND Customer
    AGGREGATE
        COUNT(*) AS total,
        COUNT(CASE WHEN annual_revenue IS NOT NULL THEN 1 END) AS has_revenue,
        COUNT(CASE WHEN tier IS NOT NULL THEN 1 END) AS has_tier
""")
print(f"Revenue completeness: {quality[0].has_revenue / quality[0].total:.1%}")
print(f"Tier completeness: {quality[0].has_tier / quality[0].total:.1%}")

#Key Takeaways

Multi-source support: Relational databases, files, APIs, message queues, and object storage all covered
Auto-discovery: Schema auto-discovery and mapping suggestions reduce manual configuration
Data validation: Validate data quality before import with pass rate thresholds
Incremental/real-time: Watermark-based incremental sync and Kafka real-time import supported
Merge strategy: Multi-source data can be upserted and merged into the same Object Type
Debug-friendly: Dry runs, previews, and debug logging help quickly troubleshoot issues

#Next Article

Next: S12-13 Subscription and Notification Guide — Learn how to configure event subscriptions and multi-channel notifications.

Tags: Data Import ETL Connectors Schema Mapping Data Validation coomia-dip