From Tb-scale Mongodb To Doris: 5 Critical Challenges And Fixes With Apache Seatunnel

Recently, in several data middle-platform projects, we have frequently used Apache SeaTunnel to synchronize data from MongoDB to Apache Doris. Honestly, this task looks simple, but once you actually get started, you realize there are quite a few pitfalls. Especially in production environments, where data volume is large and structures are complex, a small oversight can easily lead to problems.

This article does not intend to repeat those basic configuration steps—there are already plenty of them online. I want to focus on the areas in real production environments where people are most likely to stumble. Particularly when you are dealing with TB-level MongoDB collections that need to be stably synchronized to Doris for real-time analysis, the following five pitfalls are almost inevitable. I will combine specific error logs, troubleshooting approaches, and the solutions our team has explored to help you eliminate these issues one by one.

1. Data Type Mapping: BSON-to-SQL Conversion Issues

MongoDB’s BSON type system and Doris’s SQL type system appear to be automatically mappable on the surface, but in reality they hide quite a few “surprises.” The most typical examples are the handling of Decimal128 and ObjectId.

1.1 Precision Loss Issue with Decimal128

MongoDB uses Decimal128 to store high-precision numeric values, such as financial transaction amounts. SeaTunnel maps it to Doris’s DECIMAL type by default, but there is a key limitation here: Doris’s DECIMAL supports a maximum precision of 38 digits, while Decimal128 supports 34 decimal digits. If you do not explicitly specify precision in the SeaTunnel schema, you are likely to encounter the following error:

java.lang.ArithmeticException: Non-terminating decimal expansion; no exact representable decimal result  

The solution is to explicitly declare the precision in the schema. Do not rely on automatic inference—control it manually:

source {  
  MongoDB {  
    uri = "mongodb://user:password@host:27017"  
    database = "finance"  
    collection = "transactions"  
    schema = {  
      fields {  
        _id = string  
        amount = "decimal(38, 18)"  # Explicitly specify total precision 38, scale 18  
        currency = string  
        timestamp = timestamp  
      }  
    }  
  }  
}  

Note: If the Decimal128 values in your data contain more than 18 decimal places, you need to adjust according to your actual situation. In one of our e-commerce projects, coupon calculation required high precision, so we used decimal(38, 24).

1.2 ObjectId and Nested Document Serialization Pitfalls

The _id field in MongoDB is of type ObjectId by default, and SeaTunnel converts it to a string. This seems fine—until you discover primary key conflicts in the Doris table—because after ObjectId is converted to a string, Doris’s UNIQUE KEY check may cause issues.

Nested documents are even more troublesome. A common structure in MongoDB looks like this:

{  
  "_id": ObjectId("507f1f77bcf86cd799439011"),  
  "user": {  
    "name": "Zhang San",  
    "address": {  
      "city": "Beijing",  
      "district": "Chaoyang"  
    }  
  }  
}  

By default, SeaTunnel converts the entire user object into a JSON string and stores it in a VARCHAR field in Doris. If you want to directly query user.address.city in Doris, you have to use JSON functions to parse it, which results in poor performance.

Our approach is to flatten nested fields in advance using the transform plugin in SeaTunnel:

transform {  
  # Flatten nested fields  
  sql {  
    query = """  
      SELECT   
        _id,  
        user.name as user_name,  
        user.address.city as city,  
        user.address.district as district  
      FROM mongodb_source  
    """  
  }  
}  
  
sink {  
  Doris {  
    fenodes = "fe1:8030,fe2:8030"  
    username = "admin"  
    password = "***"  
    database = "analytics"  
    table = "user_flat"  
    # The table structure is now flat, and query performance is high  
  }  
}  

If the nesting level is too deep or uncertain, you may also consider using the MAP type in Doris, but note that it is only supported in version 2.0 and above.

2. Connection and Timeout Configuration: High-Concurrency Challenges in Production

In a test environment with dozens of records, everything works fine. Once you move to production, connection timeouts, cursor timeouts, and memory overflows all appear.

2.1 MongoDB Connection Pool and Cursor Timeout

Several key parameters in the MongoDB source plugin of SeaTunnel are easy to overlook:

We once encountered a major pitfall: setting cursor.no-timeout=true together with large data queries caused hundreds of cursors to accumulate on the MongoDB server side, each consuming memory and nearly bringing the cluster down. Later, we changed it to:

source {  
  MongoDB {  
    uri = "mongodb://user:password@host1:27017,host2:27017/?replicaSet=rs0&readPreference=secondaryPreferred"  
    database = "logs"  
    collection = "access_logs"  
    cursor.no-timeout = false  
    fetch.size = 16384  
    max.time-min = 30  
    partition.split-key = "_id"  
    partition.split-size = 1048576  # 1MB per partition  
  
    # Only synchronize data from the last 7 days to avoid full table scans  
    match.query = "{timestamp: {$gte: ISODate('2024-01-01T00:00:00Z')}}"  
  }  
}  

2.2 Stream Load Tuning in Doris

On the Doris Sink side, the core lies in Stream Load batch parameters. The default configuration is friendly for small data volumes, but production environments require tuning:

sink {  
  Doris {  
    fenodes = "fe1:8030,fe2:8030,fe3:8030"  
    username = "sync_user"  
    password = "***"  
    database = "dw"  
    table = "fact_table"  
  
    sink.label-prefix = "seatunnel_sync"  
    sink.enable-2pc = true  # Enable two-phase commit to ensure Exactly-Once  
    sink.buffer-size = 524288  # 512KB, default 256KB is too small  
    sink.buffer-count = 5  # Number of buffers  
    doris.batch.size = 5000  # 5000 rows per batch, default 1024  
  
    # Key: advanced Stream Load parameters  
    doris.config = {  
      format = "json"  
      read_json_by_line = "true"  
      strip_outer_array = "true"  
      num_as_string = "true"  # Convert numbers to string to avoid type issues  
  
      # Connection and timeout control  
      connect_timeout = "10"  
      socket_timeout = "30"  
  
      # Partial update mode (if the table uses Unique model)  
      partial_columns = "true"  
      merge_type = "MERGE"  
    }  
  }  
}  

There is a detail here: sink.label-prefix must be unique for each task; otherwise, Doris will reject duplicate import labels. We use the pattern "seatunnel_${job_id}_${timestamp}".

3. Performance Bottleneck Identification and Optimization: From Hours to Minutes

When synchronization tasks run slowly, it is usually not caused by a single reason, but by multiple factors stacking together.

3.1 Diagnostic Toolchain

First, you need to know where the bottleneck is. Our commonly used monitoring combination:

1. SeaTunnel’s own logs: enable DEBUG level to see the reading progress of each partition
2. MongoDB Profiler: temporarily enable to confirm whether the query uses indexes
3. Doris FE/BE monitoring: use show proc '/current_queries' to check import status
4. System monitoring: CPU, memory, network IO

There was once a case where synchronization speed was stuck at 1000 records per second and could not improve. After investigation, we found:

• On the MongoDB side: the query used the $or operator and did not use an index
• Network: cross-availability-zone transmission with high latency
• On the Doris side: BE node disk IO was saturated

3.2 Partition Strategy Optimization

SeaTunnel supports parallel reading based on partition.split-key. However, using _id as the default partition key is not always optimal.

If the data has a natural time dimension, such as a log table, partitioning by a time field works better:

source {  
  MongoDB {  
    # Assume each record has an event_time field  
    partition.split-key = "event_time"  
    partition.split-size = 3600000  # Partition by 1 hour  
  
    # Combine with query conditions to avoid full table scans  
    match.query = """  
      {  
        event_time: {  
          $gte: ISODate("2024-01-01T00:00:00Z"),  
          $lt: ISODate("2024-01-02T00:00:00Z")  
        }  
      }  
    """  
  }  
}  

If the data distribution is uneven, you can first analyze key distribution using an aggregation query:

// Execute in MongoShell  
db.collection.aggregate([  
  { $bucketAuto: { groupBy: "$shard_key", buckets: 10 } }  
])  

3.3 Memory and GC Tuning

SeaTunnel is based on the JVM, and GC issues are common with large data volumes. Our production JVM parameters:

# seatunnel_env.sh or startup script  
export JAVA_OPTS="-Xmx8g -Xms8g \  
-XX:+UseG1GC \  
-XX:MaxGCPauseMillis=200 \  
-XX:InitiatingHeapOccupancyPercent=35 \  
-XX:ParallelGCThreads=4 \  
-XX:ConcGCThreads=2 \  
-XX:+AlwaysPreTouch \  
-XX:+UseStringDeduplication \  
-XX:+PrintGCDetails \  
-XX:+PrintGCDateStamps \  
-Xloggc:/var/log/seatunnel/gc.log"  

The key point is -XX:+AlwaysPreTouch, which pre-allocates memory at startup to avoid runtime jitter.

4. Data Consistency and Error Handling: Exactly-Once Implementation Details

Data synchronization must not lose data, nor duplicate data. SeaTunnel supports Exactly-Once semantics, but it must be configured correctly.

4.1 Pitfalls of Two-Phase Commit (2PC)

Setting sink.enable-2pc = true in the Doris Sink enables two-phase commit, theoretically ensuring Exactly-Once. However, we once encountered a strange issue: after a task failed and retried, data was duplicated.

The reason was repeated use of labels. When SeaTunnel retries after failure using the same label-prefix, Doris may treat it as the same import task and skip certain data.

Solution: include timestamp and attempt count in the label:

sink {  
  Doris {  
    sink.label-prefix = "sync_${table_name}_${now()}_${attempt_num}"  
    sink.enable-2pc = true  
    sink.max-retries = 3  
    sink.check-interval = 5000  # Check every 5 seconds  
  }  
}  

4.2 Dirty Data and Type Conversion Errors

MongoDB is schema-less; the same field may be a string in one row and a number in another. Doris has a strict schema and will report errors on type mismatch.

The needs_unsupported_type_casting parameter in SeaTunnel can help:

sink {  
  Doris {  
    # Attempt to automatically convert incompatible types, e.g., Decimal to Double  
    needs_unsupported_type_casting = true  
  
    # But it is recommended to handle this in the transform layer  
  }  
}  
  
transform {  
  # Normalize types before writing  
  sql {  
    query = """  
      SELECT   
        CAST(amount AS DOUBLE) as amount_double,  
        COALESCE(name, '') as name_safe,  # Handle null  
        REGEXP_REPLACE(description, '[\\x00-\\x1F]', '') as description_clean  
      FROM source_table  
    """  
  }  
}  

4.3 Resume from Breakpoint and Checkpoint

SeaTunnel supports Checkpoint, but the storage backend must be configured correctly. We use HDFS:

env {  
  execution.parallelism = 8  
  job.mode = "BATCH"  
  
  # Checkpoint configuration  
  checkpoint.interval = 60000  # Once per minute  
  checkpoint.timeout = 600000  # 10-minute timeout  
  checkpoint.max-concurrent-checkpoints = 1  
  
  state.backend = "hdfs"  
  state.checkpoints.dir = "hdfs://namenode:8020/seatunnel/checkpoints"  
  state.savepoints.dir = "hdfs://namenode:8020/seatunnel/savepoints"  
  
  # Restore from the latest checkpoint after task failure  
  execution.savepoint-restore.enabled = true  
}  

A detail here: if the Checkpoint frequency is too high, it affects performance; if too low, too much data may be reprocessed during recovery. We generally decide based on data volume—for example, one Checkpoint per one million rows processed.

5. Operations Monitoring and Alerting: From Passive Firefighting to Proactive Prevention

This last one is not a technical pitfall, but it is more fatal than technical ones—lack of monitoring. Often, you only discover that the sync task has failed long ago when users report incorrect data.

5.1 Key Metric Monitoring

We monitor the following metrics in Prometheus (exposed via SeaTunnel JMX):

Grafana dashboard configuration example:

-- Sync lag monitoring  
SELECT   
  time_bucket('1m', timestamp) as time,  
  source_max_timestamp - sink_max_timestamp as lag_seconds  
FROM (  
  -- Maximum timestamp from source  
  SELECT MAX(event_time) as source_max_timestamp  
  FROM mongodb_source_table  
  WHERE event_time > now() - interval '1 hour'  
) source,  
(  
  -- Maximum timestamp from sink  
  SELECT MAX(event_time) as sink_max_timestamp    
  FROM doris_target_table  
  WHERE event_time > now() - interval '1 hour'  
) sink  
GROUP BY 1  
ORDER BY 1 DESC  

5.2 Automated Repair Scripts

Some common errors can be automatically fixed. For example, insufficient Doris table space:

#!/bin/bash  
# auto_extend_doris.sh  
  
ERROR_LOG=$1  
TABLE_NAME=$(grep -o "table [a-zA-Z0-9_]*" "$ERROR_LOG" | head -1 | cut -d' ' -f2)  
  
if [[ -n "$TABLE_NAME" ]]; then  
  # Check table partition usage  
  USAGE=$(mysql -h doris-fe -P 9030 -u admin -p'***' -e \  
    "SHOW PARTITIONS FROM $TABLE_NAME WHERE UsedPercent > 90;" | wc -l)  
  
  if [[ $USAGE -gt 0 ]]; then  
    # Automatically add partition  
    mysql -h doris-fe -P 9030 -u admin -p'***' <<EOF  
    ALTER TABLE $TABLE_NAME ADD PARTITION p_$(date +%Y%m%d)   
    VALUES [("$(date +%Y-%m-%d)"), ("$(date -d '+7 days' +%Y-%m-%d)"));  
EOF  
    echo "Automatic partition extension completed, restart SeaTunnel task"  
    systemctl restart seatunnel-worker  
  fi  
fi  

5.3 Data Quality Validation

Automatically validate after synchronization completes:

# validate_sync.py  
import pymongo  
import pymysql  
from datetime import datetime, timedelta  
  
def validate_counts():  
    # MongoDB count  
    mongo_client = pymongo.MongoClient("mongodb://host:27017")  
    mongo_count = mongo_client.db.collection.count_documents({  
        "update_time": {"$gte": datetime.utcnow() - timedelta(hours=1)}  
    })  
  
    # Doris count    
    doris_conn = pymysql.connect(host="doris-fe", port=9030,  
                                 user="admin", password="***", database="dw")  
    with doris_conn.cursor() as cursor:  
        cursor.execute("""  
            SELECT COUNT(*)   
            FROM target_table   
            WHERE update_time >= DATE_SUB(NOW(), INTERVAL 1 HOUR)  
        """)  
        doris_count = cursor.fetchone()[0]  
  
    # Allow 1% deviation (considering deletes, updates, etc.)  
    diff_ratio = abs(mongo_count - doris_count) / max(mongo_count, 1)  
  
    if diff_ratio > 0.01:  
        send_alert(f"Data inconsistency: MongoDB={mongo_count}, Doris={doris_count}, Difference={diff_ratio:.2%}")  
        return False  
    return True  

After building this monitoring system, our team has never again been woken up by midnight alarms—not because there are no problems, but because issues are automatically handled before impacting the business.

These weren’t theoretical issues — they cost us real time in production. Data synchronization is not just about correct configuration; the real challenge lies in stable operation in production environments. Next time you encounter SeaTunnel synchronization issues, start troubleshooting from these five aspects—you will most likely find the direction. Every environment has its own particularities, but the core problem-solving ideas are universal.