# Adiom Welcome to [Adiom](https://adiom.io) documentation! Here you'll get an overview of Adiom's data migration tool **dsync** and all the amazing features it offers. Data mobility is becoming increasingly important as organizations today rely on cloud services and hybrid IT environments more than ever. Yet, **the process of data migration is as unpredictable as it poses serious risks**, such as data loss and downtime that can cause disruptions in business operations. The problem gets exponentially worse for multi-TB live production deployments. Traditionally, database users have a few options available for database migrations: 1\) **Offline export/import** - a no-go for any non-trivial production database due to the time it takes and a long downtime window; 2\) **Vendor tooling** - while free, it's highly limited in functionality and technical support, and caters only to lower 50% percentile of customers; 3\) **Generic SaaS ETL tools** - often slow due to their generic nature, and require manual coordination between initial sync and CDC for live migration if at all possible; their business model is charge per row processed, which makes them overly expensive for large migrations; 4\) **Manual scripting /** **Kafka** - a non-trivial commitment of senior developer resources that usually results in missed timelines and a brittle solution that took too long to build. At Adiom, we believe a better way is possible. *Our main goal is to reimagine database migration and replication as a fast, repeatable and reliable process that gives you confidence to cut-over.* {% embed url="" %} 100TB live migration from HBase (NoSQL database) to MongoDB API {% endembed %} {% hint style="info" %} Book a **free**, no strings attached, migration assessment call here: {% endhint %} ### What is Dsync? [Dsync](https://github.com/adiom-data/dsync/) is data migration and replication tool between databases, primarily focusing on relational and NoSQL databases. But we also support other data stores, like S3 and vector databases. Common use cases include: * From on-premise MongoDB to MongoDB Atlas or Azure Cosmos DB * Azure Cosmos DB to MongoDB Atlas * DynamoDB to MongoDB Atlas or Cosmos DB NoSQL * Oracle to MongoDB or Document DB * PostgreSQL to MongoDB or Document DB * SQL Server to Cosmos DB NoSQL Given source and destination databases, dsync completes an initial sync transferring all data from the source to the destination database. After the initial sync, it continuously monitors the source database for any changes and updates the destination database accordingly. Dsync can be used for one-time data migration or for continuous replication. It features real-time replication, resumability, security and embedded data validation checks. ### What makes Dsync special? 1. **Purpose-built for Live Data Migration**: Dsync is specialized for **no-downtime** **database migration and real-time replication** between NoSQL and RDBMS in homogeneous and heterogeneous scenarios. It's designed by and for developers and DevOps teams handling complex production workloads. 2. **Open-source and flexible**: Though we have the Enterprise version, Dsync is fully open-source at its core, offering greater control and flexibility without vendor lock-in. We architected Dsync to be extendable so that we can cater to extremely particular configurations that our customers might have. We support data transformations and data masking. 3. **No external infrastructure**: Dsync only uses compute - it doesn’t require storage, additional cloud infrastructure or services, making it lighter and faster to deploy compared to off-the-shelf or SaaS ETL tools that often carry external dependencies and rely on third-party cloud services for execution. 4. **Performance and Scalability**: Dsync is optimized for **high-performance migrations** with minimal downtime. It can handle large datasets and mission-critical operations 10x-100x faster than most ETL tools, which may struggle with massive data volumes in real-time, low-latency environments. 5. **Security and Compliance**: Dsync fully runs in your VPC. It doesn’t store data or send it to external systems, ensuring tighter control over data security and compliance. 6. **Resiliency and Recovery**: Dsync automatically resumes from interruptions at any stage and features data integrity checks, ensuring reliable data consistency, which is critical for enterprises. 7. **Ease of use:** Dsync is intuitive to use via CLI or Web Interface with automatic and seamless transition between initial data copy and CDC. The migration or replication process is easily controlled and observable through Dsync interface or OTel metrics. ### Distribution * [Single-binary Open Source Dsync](https://github.com/adiom-data/dsync/). Best for development and small-to-medium migrations less than few 100 GB. It is distributed under AGPL v3. * [Enterprise Dsynct.](/enterprise/scalable-deployment) Best for medium-to-large production use cases, from 100 GB to 100 TB and beyond. Offers horizontal scalability, observability and additional reliability guarantees. Distributed under commercial license with free trial. We offer support and technical services for both products. Please reach out to us at for commercial inquiries. Now that you've learned a bit about what Adiom does... ### Jump right in!
QuickstartSet up dsync/files/rYYUBydeyvtnvjhIh8BO/pages/CyH2xJQs9yWJ1S8BYNav
What is SupportedFind out which applications are supported/files/tc5z3w3ubga7Mxgxb8mw/pages/qdpM5cCieCXVNGmT28bx
Explore FeaturesLearn about dsync's key features/files/v9HxA7GIK9zkidh7DbBm/pages/cEHgDv7NGBquDGHeLyQ4
# Quickstart Learn how to set up dsync and move terabytes of data in just a few steps! # From Cosmos DB to MongoDB Follow these simple instructions to get dsync up and running and perform a migration from our Cosmos DB demo instance to your MongoDB destination. ### Step 1: Download dsync {% hint style="info" %} Working on a large-scale production environment? Use our horizontally scalable [Enterprise offering.](/enterprise/scalable-deployment) {% endhint %} Use Docker (`markadiom/dsync`) or download the latest release from the [GitHub Releases](https://github.com/adiom-data/dsync/releases/latest) page. Note that on Mac devices you may need to configure a security exception to execute the binary by following [these steps](https://support.apple.com/en-ca/guide/mac-help/mh40616/mac). Alternatively, you can build dsync from the source code. ```bash git clone https://github.com/adiom-data/dsync.git cd dsync go build ``` {% hint style="info" %} You can use Homebrew to easily install Dsync on your Mac: ``` brew install adiom-data/homebrew-tap/dsync ``` {% endhint %} *** ### Step 2: Prepare the destination MongoDB instance {% hint style="warning" %} If you already have the desired destination MongoDB instance up and running, you can skip this step. {% endhint %} 1. Install [MongoDB](https://www.mongodb.com/docs/manual/administration/install-community/) 2. Start a local MongoDB instance: ```bash mkdir ~/temp cd ~/temp mkdir data_d mongod --dbpath data_d --logpath mongod_d.log --fork --port 27017 ``` *** ### Step 3: Start dsync ```bash # You can use our publicly accessible (read-only) Cosmos DB instance export COSMOS_DEMO=$(echo bW9uZ29kYjovL2Nvc21vc2RiLWRlbW8taW5zdGFuY2U6SkhiRWpRb2JkWm03YWJEcFp2UzZrWHpBMDRXNTBJd2V4QmlQYnVJWFQ2TElmNkhsV2V4YWphQzhkd042REJ2YVh6ajBnclFrdkwzY0FDRGJONjdxZWc9PUBjb3Ntb3NkYi1kZW1vLWluc3RhbmNlLm1vbmdvLmNvc21vcy5henVyZS5jb206MTAyNTUvP3NzbD10cnVlJnJlcGxpY2FTZXQ9Z2xvYmFsZGImcmV0cnl3cml0ZXM9ZmFsc2UmbWF4SWRsZVRpbWVNUz0xMjAwMDAmYXBwTmFtZT1AY29zbW9zZGItZGVtby1pbnN0YW5jZUA= | base64 --decode) # Feel free to use your own MongoDB connection string export MDB_DEST='mongodb://localhost:27017' ./dsync --progress --logfile dsync.log $COSMOS_DEMO $MDB_DEST ``` Now Dsync should be running! Feel free to interrupt the sync process (via Ctrl+C) it once the initial sync is done. The demo database has a few million records and the resources are shared - allow 5-10 minutes for the process to complete. *** ### Step 4: Check the data ```bash mongosh $MDB_DEST ``` Congratulations! You should be able to access the 'odc' database and see the collections in it that were migrated from the Cosmos DB. # From DynamoDB to MongoDB Near-zero downtime migration from DynamoDB to MongoDB with dsync ## Prerequisites 1\) DynamoDB instance with [change streams enabled](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html) (if CDC is needed). Make sure you have at least "New Image" enabled for the stream. 2\) MongoDB cluster. Any sharded databases and collections need to be pre-created. 3\) AWS credentials with proper permissions for the source 4\) Installed [aws-cli](https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html) or AWS\_ACCESS\_KEY\_ID and AWS\_SECRET\_ACCESS\_KEY ### Additional Considerations #### Indexes For faster migration performance, we recommend to create secondary indexes on destination MongoDB after the initial sync is done. All modern MongoDB versions support building indexes in the background, and they can be created during the CDC / catch up phase. #### Networking The host running dsync should have network access to both DynamoDB and MongoDB instances. #### Provisioning additional capacity on the source The initial data copy stage of the migration is equivalent to a parallel table scan in DynamoDB and it will consume additional read units. When your DynamoDB source is already serving live production traffic and the table(s) are configured with [provisioned capacity](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/provisioned-capacity-mode.html), we recommended to *temporarily increase read capacity* to avoid impacting production traffic. The exact capacity increase varies on a case-by-case basis, but as a rule of thumb we recommend to add at least 10,000 RCU. The consumed capacity can be regulated using the `--load-level [Low|Medium|High]` dsync command-line parameter or via more granular `--parallel-copiers N` option.
Example Each copier or reader thread scans data in 1MB pages (256 RCU each), and can process 5 pages per second. With record size of 2KB, we get 500 records per page, or 2500/s per copier. With the default 4 copiers, we should be able to achieve 10,000 records per second consuming 5,120 RCU.
{% hint style="info" %} Note that DynamoDB uses *adaptive scaling* where maximum RCU per *adaptive partition* is 3,000. When needed (and possible), adaptive partitions are automatically split based on the partition key. This splitting can lead to temporary throttling exceptions. Dsync does its best to avoid them, but in case they happen, the associated initial sync tasks are safe to retry. {% endhint %} ## Step 1: Download dsync [![Adiom on Azure Marketplace](https://assetsprod.microsoft.com/mpn/en-us/ms-azure-marketplace.png)](https://azuremarketplace.microsoft.com/en/marketplace/apps/adiom.adiom_dsync_vm_dynamo) {% hint style="info" %} Working on a large-scale production environment? Use our horizontally scalable [Enterprise offering.](/enterprise/scalable-deployment) {% endhint %} Use Docker (`markadiom/dsync`) or download the latest release from the [GitHub Releases](https://github.com/adiom-data/dsync/releases/latest) page. Note that on Mac devices you may need to configure a security exception to execute the binary by following [these steps](https://support.apple.com/en-ca/guide/mac-help/mh40616/mac). You can also build dsync directly from the [source code](https://github.com/adiom-data/dsync/) using `go build`. If you're using a Cloud Provider marketplace image (e.g. from [Azure Marketplace](https://azuremarketplace.microsoft.com/en/marketplace/apps/adiom.adiom_dsync_vm_dynamo?tab=Overview)), then the binaries have already been preinstalled, and you just need to ssh into your provisioned instance. If you want to access the Web UI progress feature (default port 8080) you can port forward e.g. ``` ssh -L 8080:localhost:8080 myuser@my.instance.ip ``` {% hint style="info" %} You can use Homebrew to easily install Dsync on your Mac: ``` brew install adiom-data/homebrew-tap/dsync ``` {% endhint %} ## Step 2: Set up environment variables Ensure you set your AWS credentials properly, such as by setting the AWS environment variables: ``` export AWS_ACCESS_KEY_ID="..." export AWS_SECRET_ACCESS_KEY="..." export AWS_SESSION_TOKEN="..." ``` Alternatively, you can use the `aws configure sso` (if you're doing it for the first time or using a VM from the marketplace) and `aws sso login` shell commands to securely login into the AWS account for aws-cli. Test to see if you can see your dynamodb tables with `aws dynamodb list-tables`. ## (Optional) Step 3: Start the transformer When data transformations are required, you can connect a transformer to Dsync via the gRPC extension interface. You can [write your own](/basics/features#transformations) in the language of your choice, or use our [YAML-based declarative transformer](/enterprise/running-dsynct/data-transformations) (*available only for **Enterprise** customers*). When running the transformer in Docker, make sure the container is on the same network as Dsync container: the `--network` option to the "docker run" command. ## Step 4: Start dsync Run `dsync --namespace :. dynamodb $MONGODB_URI`. Replace `` with the dynamodb table name. Replace `$MONGODB_URI` with the desired MongoDB URI. When transformer is required, use `dsync --namespace :. dynamodb $MONGODB_URI grpc://localhost:8085 --insecure` We use the `--insecure` since we are not using TLS for our connection to the transformer service and we assume it's running on the same host on port 8085. You can migrate multiple different tables at the same time by specifying multiple mappings in the `--namespace` param: `dsync --namespace ":.,:." dynamodb $MONGODB_URI grpc://localhost:8089 --insecure` For Cloud Marketplace images and Docker: ``` sudo docker run \ --network mynet --name dsync \ -p 8080:8080 \ -e AWS_ACCESS_KEY_ID=$AWS_ACCESS_KEY_ID \ -e AWS_SECRET_ACCESS_KEY=$AWS_SECRET_ACCESS_KEY \ -e AWS_REGION=$AWS_REGION \ markadiom/dsync \ --web-host 0.0.0.0 \ --namespace ":.,:." \ dynamodb \ $MONGODB_URI \ grpc://transformer:8085 --insecure ``` ## Step 5: Post-migration configuration #### Indexes Create/validate necessary [indexes](https://www.mongodb.com/docs/manual/indexes/) on MongoDB. # From self-managed MongoDB to Azure DocumentDB Follow these simple instructions to get dsync up and running and perform a migration from your self-managed MongoDB to an Azure DocumentDB or Cosmos DB destination. ### Step 1: Download dsync [![Adiom on Azure Marketplace](https://assetsprod.microsoft.com/mpn/en-us/ms-azure-marketplace.png)](https://azuremarketplace.microsoft.com/en-us/marketplace/apps/adiom.adiom_dsync_vm) {% hint style="info" %} Working on a large-scale production environment? Use our horizontally scalable [Enterprise offering.](/enterprise/scalable-deployment) {% endhint %} Use Docker (`markadiom/dsync`) or download the latest release from the [GitHub Releases](https://github.com/adiom-data/dsync/releases/latest) page. Note that on Mac devices you may need to configure a security exception to execute the binary by following [these steps](https://support.apple.com/en-ca/guide/mac-help/mh40616/mac). Alternatively, you can build dsync from the source code. ```bash git clone https://github.com/adiom-data/dsync.git cd dsync go build ``` {% hint style="info" %} You can use Homebrew to easily install Dsync on your Mac: ``` brew install adiom-data/homebrew-tap/dsync ``` {% endhint %} *** ### Step 2: Prepare the source MongoDB instance {% hint style="warning" %} If you already have the desired source MongoDB instance up and running, you can skip this step. {% endhint %} 1. Install [MongoDB](https://www.mongodb.com/docs/manual/administration/install-community/) 2. Start a local MongoDB instance 3. Load sample data ```bash # Start a new mongod instance on localhost:27017 with dbpath ~/temp/data_d mkdir ~/temp cd ~/temp mkdir data_d mongod --dbpath data_d --logpath mongod_d.log --fork --port 27017 # Load sample data git clone https://github.com/mcampo2/mongodb-sample-databases cd mongodb-sample-databases mongorestore dump/sample_mflix ``` *** ### Step 3: Prepare the destination DocumentDB instance {% hint style="warning" %} If you already have the desired source Cosmos DB instance up and running, you can skip this step. {% endhint %} Follow official Azure documentation to set up a DocumentDB instance. For example, these are the instructions for setting up Azure DocumentDB (formerly Cosmos DB vCore for MongoDB) instance using Azure Portal: Make sure to note the DocumentDB cluster MongoDB connection string: *** ### Step 3: Start dsync ```bash # Use the read-write connection string from the Azure Portal export COSMOS_DEST= # Feel free to use your own MongoDB connection string export MDB_SRC='mongodb://localhost:27017' # Set the mode to InitialSync if the source MongoDB is NOT a replica set # Otherwise skip the option altogether or set it to Full to switch to CDC after Initial Sync ./dsync --mode InitialSync --progress --logfile dsync.log $MDB_SRC $COSMOS_DEST ``` Now Dsync should be running! Feel free to interrupt the sync process (via Ctrl+C) it once the initial sync is done. The sample database has a few million records - allow 5-10 minutes for the process to complete. *** ### Step 4: Check the data ```bash mongosh $COSMOS_DEST ``` Congratulations! You should be able to access the 'sample\_mflix' database and see the collections in it that were migrated from the local MongoDB instance. *** ### Contact us Please reach out to us [by email](mailto:info@adiom.io) and on [Discord](https://discord.gg/r4xzVfMQeU) for commercial inquiries and support. # From DynamoDB to Cosmos DB NoSQL Near-zero downtime migration from DynamoDB to Cosmos DB NoSQL with dsync ## Prerequisites 1\) DynamoDB instance with [change streams enabled](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html). Make sure you have at least "New Image" enabled for the stream. 2\) Cosmos DB NoSQL Account with the pre-created destination database(s) and container(s) 3\) AWS credentials with proper permissions for the source 4\) CosmosDB NoSQL Account Url and Read-Write Key (can be obtained using the Azure Portal) 5\) Installed [aws-cli](https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html) or AWS\_ACCESS\_KEY\_ID and AWS\_SECRET\_ACCESS\_KEY ### Additional Considerations #### Indexes For faster migration performance, we recommend to disable indexes on Cosmos DB NoSQL tables that will be used as a destination for the migration - that gives 100-150% performance boost for bulk insertion. #### Networking The host running dsync should have network access to both DynamoDB and Cosmos DB for NoSQL instances. When both DynamoDB and Cosmos DB are only accessible from within a VPC/VNET, you need to establish connectivity between the cloud providers by following the instructions [here](https://learn.microsoft.com/en-us/azure/cloud-adoption-framework/ready/azure-best-practices/connectivity-to-other-providers). Note that Cosmos DB uses dynamic ports as explained in the [documentation](https://learn.microsoft.com/en-us/azure/cosmos-db/nosql/sdk-connection-modes#service-port-ranges). #### Provisioning additional capacity on the source The initial data copy stage of the migration is equivalent to a parallel table scan in DynamoDB and it will consume additional read units. When your DynamoDB source is already serving live production traffic and the table(s) are configured with [provisioned capacity](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/provisioned-capacity-mode.html), we recommended to *temporarily increase read capacity* to avoid impacting production traffic. The exact capacity increase varies on a case-by-case basis, but as a rule of thumb we recommend to add at least 10,000 RCU. The consumed capacity can be regulated using the `--load-level [Low|Medium|High]` dsync command-line parameter or via more granular `--parallel-copiers N` option.
Example Each copier or reader thread scans data in 1MB pages (256 RCU each), and can process 5 pages per second. With record size of 2KB, we get 500 records per page, or 2500/s per copier. With the default 4 copiers, we should be able to achieve 10,000 records per second consuming 5,120 RCU.
{% hint style="info" %} Note that DynamoDB uses *adaptive scaling* where maximum RCU per *adaptive partition* is 3,000. When needed (and possible), adaptive partitions are automatically split based on the partition key. This splitting can lead to temporary throttling exceptions. Dsync does its best to avoid them, but in case they happen, the associated initial sync tasks are safe to retry. {% endhint %} ## Step 1: Download dsync [![Adiom on Azure Marketplace](https://assetsprod.microsoft.com/mpn/en-us/ms-azure-marketplace.png)](https://azuremarketplace.microsoft.com/en/marketplace/apps/adiom.adiom_dsync_vm_dynamo) {% hint style="info" %} Working on a large-scale production environment? Use our horizontally scalable [Enterprise offering.](/enterprise/scalable-deployment) {% endhint %} Use Docker (`markadiom/dsync`) or download the latest release from the [GitHub Releases](https://github.com/adiom-data/dsync/releases/latest) page. Note that on Mac devices you may need to configure a security exception to execute the binary by following [these steps](https://support.apple.com/en-ca/guide/mac-help/mh40616/mac). You can also build dsync directly from the [source code](https://github.com/adiom-data/dsync/) using `go build`. If you're using a Cloud Provider marketplace image (e.g. from [Azure Marketplace](https://azuremarketplace.microsoft.com/en/marketplace/apps/adiom.adiom_dsync_vm_dynamo?tab=Overview)), then the binaries have already been preinstalled, and you just need to ssh into your provisioned instance. If you want to access the Web UI progress feature (default port 8080) you can port forward e.g. ``` ssh -L 8080:localhost:8080 myuser@my.instance.ip ``` {% hint style="info" %} You can use Homebrew to easily install Dsync on your Mac: ``` brew install adiom-data/homebrew-tap/dsync ``` {% endhint %} ### CosmosDB NoSQL Sink Binary The connector for CosmosDB NoSQL runs as a separate process because it uses Java SDK optimized for bulk operations. If you're using a Cloud Provider marketplace image (e.g. from Azure Marketplace), you will be using a Docker image to run the connector. If not, you can check out the git repository, `cd` into the `java` directory and run `mvn clean install`. You will need Java JDK 21 or newer. This will create a jar in the `java/target` directory and for convenience you can set up an alias like so (replacing the path/to/dsync with the appropriate file): ``` alias cosmos-sink='OTEL_SDK_DISABLED=true java -jar /path/to/dsync/java/target/cosmos-connector-1-jar-with-dependencies.jar' ``` You can look at the README in the `java` directory for the most up to date set up instructions. ## Step 2: Set up environment variables 1. Look up the target CosmosDB NoSQL Account url and key and export the details into the env variables $URL and $KEY respectively. Ensure you create the database and container you want to move files into. 2. Ensure you set your AWS credentials properly, such as by setting the AWS environment variables: ``` export AWS_ACCESS_KEY_ID="..." export AWS_SECRET_ACCESS_KEY="..." export AWS_SESSION_TOKEN="..." ``` Alternatively, you can use the `aws configure sso` (if you're doing it for the first time or using a VM from the marketplace) and `aws sso login`shell commands to securely login into the AWS account for aws-cli. Test to see if you can see your dynamodb tables with `aws dynamodb list-tables`. ## Step 3: Start the Cosmos NoSQL connector You will need to set the URL and the KEY env variables to the correct values corresponding to yoru Cosmos DB account. See [here](https://learn.microsoft.com/en-us/answers/questions/1056745/where-to-find-cosmos-db-endpoint-and-key) for where to find them. ``` export URL="..." export KEY="..." ``` Then run `cosmos-sink 8089 $URL $KEY &` in the background. This starts a grpc service (running without tls) that will write to the specified CosmosDB NoSQL destination. If you're building dsync from the source, follow the instructions [here](https://github.com/adiom-data/dsync/blob/main/java/README.md) to build the connector. For Cloud Marketplace images and Docker: ``` sudo docker network create mynet sudo docker run -d \ --network mynet --name cosmosnosqlconnector \ -e OTEL_SDK_DISABLED=true \ markadiom/cosmosnosqlconnector 8089 $URL $KEY ``` Use `sudo docker logs -f cosmosnosqlconnector` to follow connector logs. ## Step 4: Start dsync Run `dsync --namespace :. dynamodb grpc://localhost:8089 --insecure`. Replace `` with the dynamodb table name. Replace `.` with the desired CosmosDB NoSQL Database and Container names. We use the `--insecure` since we are not using TLS for our connection to the Cosmos DB NoSQL connector. You can migrate multiple different tables at the same time by specifying multiple mappings in the `--namespace` param: `dsync --namespace ":.,:." dynamodb grpc://localhost:8089 --insecure` For Cloud Marketplace images and Docker: ``` sudo docker run \ --network mynet --name dsync \ -p 8080:8080 \ -e AWS_ACCESS_KEY_ID=$AWS_ACCESS_KEY_ID \ -e AWS_SECRET_ACCESS_KEY=$AWS_SECRET_ACCESS_KEY \ -e AWS_REGION=$AWS_REGION \ markadiom/dsync \ --web-host 0.0.0.0 \ --namespace ":.,:." \ dynamodb \ grpc://cosmosnosqlconnector:8089 --insecure ``` ## Step 5: Post-migration configuration #### Indexes Create/validate necessary [indexes](https://learn.microsoft.com/en-us/azure/cosmos-db/index-policy) on Cosmos DB for NoSQL. #### Global tables If you had global tables configured in DynamoDB, you may want to configure [global distribution](https://learn.microsoft.com/en-us/azure/cosmos-db/distribute-data-globally) for those tables in Cosmos DB for NoSQL. #### Reporting Follow these [instructions](https://learn.microsoft.com/en-us/azure/cosmos-db/analytics-and-business-intelligence-overview) to configure Analytics and BI in Azure for data stored in Cosmos DB for NoSQL. #### Monitoring Familiarize yourself with Cosmos DB [monitoring dashboards and metrics](https://learn.microsoft.com/en-us/azure/cosmos-db/monitor?tabs=resource-specific-diagnostics). You can configure the necessary alerts by following the instructions [here](https://learn.microsoft.com/en-us/azure/cosmos-db/create-alerts). #### Backups [Review and adjust configuration](https://learn.microsoft.com/en-us/azure/cosmos-db/online-backup-and-restore) for Cosmos DB backups. ## Limitations 1. The DynamoDB to Cosmos DB for NoSQL data flow is currently not resumable. 2. Embedded validation checks may not function for this data flow. # From Cosmos DB to /dev/null ### Step 1: Download dsync Use Docker (`markadiom/dsync`) or download the latest release from the [GitHub Releases](https://github.com/adiom-data/dsync/releases/latest) page. Note that on Mac devices you may need to configure a security exception to execute the binary by following [these steps](https://support.apple.com/en-ca/guide/mac-help/mh40616/mac). Alternatively, you can build dsync from the source code. 
git clone https://github.com/adiom-data/dsync.git
cd dsync
go build
{% hint style="info" %} You can use Homebrew to easily install Dsync on your Mac: ``` brew install adiom-data/homebrew-tap/dsync ``` {% endhint %} *** ### Step 2: Start dsync 
# You can use our publicly accessible (read-only) Cosmos DB instance
export COSMOS_DEMO=$(echo bW9uZ29kYjovL2Nvc21vc2RiLWRlbW8taW5zdGFuY2U6SkhiRWpRb2JkWm03YWJEcFp2UzZrWHpBMDRXNTBJd2V4QmlQYnVJWFQ2TElmNkhsV2V4YWphQzhkd042REJ2YVh6ajBnclFrdkwzY0FDRGJONjdxZWc9PUBjb3Ntb3NkYi1kZW1vLWluc3RhbmNlLm1vbmdvLmNvc21vcy5henVyZS5jb206MTAyNTUvP3NzbD10cnVlJnJlcGxpY2FTZXQ9Z2xvYmFsZGImcmV0cnl3cml0ZXM9ZmFsc2UmbWF4SWRsZVRpbWVNUz0xMjAwMDAmYXBwTmFtZT1AY29zbW9zZGItZGVtby1pbnN0YW5jZUA= | base64 --decode)

# Use /dev/null as the destination
export DEST='/dev/null' 

./dsync --progress --logfile dsync.log $COSMOS_DEMO $DEST
Now dsync should be running! Feel free to interrupt the sync process (via Ctrl+C) it once the initial sync is done. The demo database has a few million records and the resources are shared - allow 5-10 minutes for the process to complete. # From Cosmos DB NoSQL to MongoDB API Near-zero downtime migration from Cosmos DB NoSQL to MongoDB API ### Prerequisites 1. **Obtain Cosmos DB credentials** - URI and Primary Key ("Settings" -> "Keys" in the Azure Portal) 2. **Obtain MongoDB connection string** - for the destination cluster 3. **Enable** [**"All Versions and Deletes"**](https://learn.microsoft.com/en-us/azure/cosmos-db/change-feed-modes?tabs=latest-version#all-versions-and-deletes-change-feed-mode-preview) for your Cosmos DB container ("Settings" -> "Features" in the Azure Portal) * If you can't turn on the feature, you can run the Cosmos DB NoSQL connector with the `COSMOS_DISABLE_ALL_VERSIONS_AND_DELETES=true` environment variable. In that case Dsync will not be able to replicate delete events from Cosmos DB. ### Data types and ID considerations Cosmos DB NoSQL uses **JSON** while MongoDB uses **BSON**, so a [data transformation](/enterprise/running-dsynct/data-transformations) is required. Commonly, as part of that transformation, you'd want to convert some JSON types into BSON types, such as strings into Dates for timestamps. Additionally, you may want to transfer some internal Cosmos NoSQL fields such as `_ts` that is used for TTL. For the Open Source dsync, you can build your own custom transformer following this [example in GitHub](https://github.com/adiom-data/dsync/blob/main/transform/identity.go). You can implement it in the language of your choice (e.g. Java or Python) as long as it supports gRPC and implements the required [Transform interface](https://docs.adiom.io/implementation-details/architecture#transform-interface). {% hint style="info" %} The [Enterprise](/enterprise/scalable-deployment) version of Dsync has a CEL-based transformer that you can try [here](https://github.com/adiom-data/public/tree/main/dsync-transform). In the instructions below we will be using its format as an example given how intuitive it is. {% endhint %} The Cosmos DB NoSQL [ID format](/reference/connectors/cosmos-db-nosql#id-format) is composed of the shard key followed by the `id` field. MongoDB uses a single `_id` field. A transform config must map between these ID formats. See [Transform Data Types](/enterprise/running-dsynct/data-types) for full details on JSON to BSON mappings. #### Simple case: shard key is `/id` When the shard key is `/id`, the Cosmos DB ID contains only the `id` field. The transform maps `id` to `_id`: ```yaml # transform.yaml defaultmapping: default mappings: - namespace: default delete: ["id"] add: ["_id"] mapid: id cel: _id: id ``` #### With a shard key prefix When the shard key is a separate field (e.g. `/region`), the Cosmos DB ID is multi-part (e.g. `["us-east", "123"]`). You need to use `idkeys` to declare the source ID fields and collapse them into a single `_id`: ```yaml # transform.yaml defaultmapping: default idlist: true mappings: - namespace: default idkeys: ["region", "id"] delete: ["region", "id"] add: ["_id"] mapid: id[1] cel: _id: id[1] ``` {% hint style="info" %} Adjust `idkeys` and the `cel` expressions to match your Cosmos DB container's shard key configuration. See the [multi-part ID examples](/enterprise/running-dsynct/data-types#multi-part-id-examples) for more patterns. {% endhint %} *** ### Step 1: Download dsync {% hint style="info" %} Working on a large-scale production environment? Use our horizontally scalable [Enterprise offering.](/enterprise/scalable-deployment) {% endhint %} Use Docker (`markadiom/dsync`) or download the latest release from the [GitHub Releases](https://github.com/adiom-data/dsync/releases/latest) page. Note that on Mac devices you may need to configure a security exception to execute the binary by following [these steps](https://support.apple.com/en-ca/guide/mac-help/mh40616/mac). Alternatively, you can build dsync from the source code. ```bash git clone https://github.com/adiom-data/dsync.git cd dsync go build ``` {% hint style="info" %} You can use Homebrew to easily install Dsync on your Mac: ``` brew install adiom-data/homebrew-tap/dsync ``` {% endhint %} {% hint style="info" %} We recommend using Docker for this tutotial {% endhint %} #### CosmosDB NoSQL Connector The connector for CosmosDB NoSQL runs as a separate process because it uses the optimized Java SDK. You can run it as a Docker container. If you'd rather build it from the source and run as a regular process, you can check out the git repository, `cd` into the `java` directory and run `mvn clean install`. You will need Java JDK 21 or newer. This will create a jar in the `java/target` directory and for convenience you can set up an alias like so (replacing the path/to/dsync with the appropriate file): ``` alias cosmos-connector='OTEL_SDK_DISABLED=true java -jar /path/to/dsync/java/target/cosmos-connector-1-jar-with-dependencies.jar' ``` You can look at the README in the `java` directory for the most up to date set up instructions. *** ### Step 2: Prepare the destination MongoDB instance {% hint style="warning" %} If you already have the desired destination MongoDB instance up and running, you can skip this step. {% endhint %} 1. Install [MongoDB](https://www.mongodb.com/docs/manual/administration/install-community/) 2. Start a local MongoDB instance: ```bash mkdir ~/temp cd ~/temp mkdir data_d mongod --dbpath data_d --logpath mongod_d.log --fork --port 27017 ``` {% hint style="info" %} For faster performance, we recommend creating any required secondary indexes *after* the initial data copy has completed. {% endhint %} Export MongoDB URI as a shell variable: {% code overflow="wrap" %} ```bash export MONGODB_URI=<...> #e.g. mongodb+srv://user:pass@cluster.mongodb.net ``` {% endcode %} *** ### Step 3: Start the Cosmos NoSQL connector You will need to set the URL and the KEY env variables to the correct values corresponding to your Cosmos DB account. See [here](https://learn.microsoft.com/en-us/answers/questions/1056745/where-to-find-cosmos-db-endpoint-and-key) for where to find them. ```bash export URL="..." export KEY="..." ``` Then run `cosmos-connector 8089 $URL $KEY &` in the background. This starts a grpc service (running without tls) that will talk to Cosmos DB NoSQL. If you're building dsync from the source, follow the instructions [here](https://github.com/adiom-data/dsync/blob/main/java/README.md) to build the connector. For Cloud Marketplace images and Docker: ```bash sudo docker network create mynet sudo docker run -d \ --network mynet --name cosmosnosqlconnector \ -e OTEL_SDK_DISABLED=true \ markadiom/cosmosnosqlconnector 8089 $URL $KEY ``` Use `sudo docker logs -f cosmosnosqlconnector` to follow connector logs. *** ### Step 4: Start the transformer You can start your transformer gRPC server listening on a port like 8085. When using the Enterprise CEL-based transformer, you will need to prepare the config file as described in [#data-types-and-id-considerations](#data-types-and-id-considerations "mention") , save it as `config.yml`, and run the process as a Docker container: {% code overflow="wrap" %} ```bash sudo docker run -d \ --network mynet --name dsync-transform \ -v "./config.yml:/config.yml" \ -e "DSYNCT_MODE=simple" \ markadiom/dsynct --host-port=0.0.0.0:8085 transformer ``` {% endcode %} Use `sudo docker logs -f dsync-transform` to follow transformer logs. *** ### Step 5: Start dsync Run `dsync --namespace . $COSMOS_NOSQL_GRPC_URI --insecure $MONGODB_URI $TRANSFORMER_GRPC_URI --insecure`. Substitute `GRPC_URI` with corresponding addresses for the connector and the transformer in the format `grpc://localhost:port` Replace `.` with the desired CosmosDB NoSQL Database and Container names. We use the `--insecure` since we are not using TLS for our connection to the Cosmos DB NoSQL connector. You can migrate multiple different containers at the same time by specifying multiple mappings in the `--namespace` param: `dsync --namespace ".,."` Full command for Docker: ```bash sudo docker run \ --network mynet --name dsync \ -p 8080:8080 \ markadiom/dsync \ --web-host 0.0.0.0 \ --namespace "." \ grpc://cosmosnosqlconnector:8089 --insecure \ $MONGODB_URI \ grpc://dsync-transform:8085 --insecure ``` The web progress will be available on [localhost:8080](https://localhost:8080). ### Limitations For Cosmos DB NoSQL sources, the Open Source version of Dsync only supports CDC for a single namespace . For multiple namespaces, you can either do the initial sync only (`--mode InitialSync`), run multiple Dsync processes (one for each namespace), or use the [Enterprise version](/enterprise/running-dsynct/cosmos-db-nosql-to-mongo). # What is supported Here's a list of our currently supported connectors. We aim to expand to other database types, bidirectional communication and complex flows soon! {% hint style="info" %} Haven't set up Dsync yet? Head to the [Quickstart](/getting-started/quickstart) section to learn more. {% endhint %} {% hint style="info" %} ***Don't see the connector you're looking for?*** [Contact us](https://www.adiom.io/contact). We can build it. {% endhint %} ## Connectors ### System
TypeVersionStatusCapabilitiesNotes
/dev/null-AvailableSourceDoes exactly what you'd expect it to do
/dev/random-AvailableSinkGenerates a stream of random operations
### NoSQL Databases
TypeVersionStatusCapabilitiesNotes
MongoDB4.2+AvailableSource, SinkSupports Atlas dedicated, Atlas serverless and self-managed installations
Cosmos DB RU (MongoDB API)4.2, 6.0+AvailableSource, SinkSupports Azure Cosmos with Mongo API with Provisioned RUs (not serverless)
Azure Document DB (Cosmos DB vCore)currentAvailableSource, SinkBoth managed and Open Source versions are supported
Cosmos DB NoSQLcurrentAvailableSource, Sink
DynamoDBcurrentAvailableSource, Sink
HBase1.x, 2.xPrivate PreviewSource, SinkIncludes CDC support
See blog for details
AWS DocumentDB4.0*, 5.0AvailableSource, Sink

Support via generic MongoDB connector.

* version 4.0 requires older Go driver.

AstraDBcurrentPrivate PreviewSource
Cassandra3.11, 4.0+Private PreviewSource, SinkFor CDC limitations see this page.
CouchDB / IBM Cloudant2.x, 3.xPrivate PreviewSource, Sink
FirestorecurrentPrivate PreviewSink
### SQL Databases / RDBMS
TypeVersionStatusCapabilitiesNotes
PostgreSQL15+AvailableSource, SinkDirect connectivity for 1:1 migrations with transformations
SQL Server / PostgreSQL / Oracle / DB2-Public PreviewSourceCustom query-based via our "SQL batch" connector
See blog and docs for details.
### Vector Databases
TypeVersionStatusCapabilitiesNotes
WeaviatelatestPublic PreviewSink
QdrantlatestIn DevelopmentSink
S3 Vector Index-Public PreviewSink
### Other
TypeVersionStatusCapabilitiesNotes
S3 Storage-AvailableSource, SinkExport into and from S3 in JSON format. CDC isn't supported: use "--mode InitialSync"
File-AvailableSource, SinkOnly CSV for now
## Metadata * **For single-binary Dsync:** MongoDB or Cosmos DB with MongoDB API * Supports local, self-managed, and Atlas * Supports any Cosmos DB with MongoDB API * **For** [**Enterprise Dsynct**](/enterprise/scalable-deployment)**:** the metadata is handled and stored by Temporal. # Compare to other tools Evaluate Dsync and alternative solutions for data migration and replication When planning a large-scale database migration or setting up real-time replication, choosing the right tool is critical. While teams often consider using traditional database utilities, cloud-native services, event streaming platforms, or generic ETL platforms, these solutions are rarely optimized for zero-downtime, production-grade cutovers. \ This guide provides an architectural breakdown of how Adiom's **Dsync** compares against common alternatives across operational, technical, and infrastructural dimensions. *** ### Architectural Comparison Matrices #### 1. Speed, Scale, and Reliability | Dimension | Adiom Dsync | Dump-Restore | Native Cloud Tools | Debezium | 3rd Party ETL | Apache Spark | | ---------------------- | ----------------------------------------- | ---------------------------------- | ---------------------- | ----------------------------------- | --------------------------------- | -------------------------------- | | **Downtime** | Sub-minute | Hours or days | Low, but unpredictable | Low | Moderate to high | High sync downtime | | **Speed** | Ultra-fast (10x-100x generic tools) | Slow | Moderate | Moderate | Slow; built for analytics batches | Fast, but expensive | | **Growth Scaling** | Horizontal or vertical scaling | None; limited by one machine | Limited cloud ceilings | Good; relies on complex Kafka setup | Bounded by vendor limitations | Excellent cluster scaling | | **Can it Resume?** | Yes, automatic checkpointing at any stage | No; must restart from scratch | Varies | Yes, via Kafka settings | Yes, via micro-batches | Requires heavy custom coding | | **One-Click Rollback** | Yes; easily reverse sync direction | No; highly complex manual recovery | Rarely supported | Requires double the pipelines | Not supported for databases | Must build reverse code manually | *** #### 2. Data Flexibility and Safety | Dimension | Adiom Dsync | Dump-Restore | Native Cloud Tools | Debezium | 3rd Party ETL | Apache Spark | | ------------------------ | ----------------------------------------------------- | ------------------------------- | ------------------------------ | ---------------------------------------- | ------------------------------------ | ----------------------------------- | | **Supported Databases** | SQL, NoSQL, S3, & Vector DBs | Same database type only | Very restrictive; cloud-locked | Mainly Relational | Good sources; weak operational sinks | Broad with manual config | | **Data Transformation** | Transformer, real-time masking and filtering built-in | None possible | Very basic mapping only | Complex Java/Kafka configurations | Extensive | Extensive, but requires high coding | | **Data Integrity Check** | Automated embedded checks | None; requires separate scripts | Basic row counts only | None; up to the application | Basic volume tracking | None; requires manual coding | | **Custom Plugins** | High; supports easy gRPC plugins | None | Low; cloud-restricted | Medium; requires custom Java development | Low; fully managed by vendor | High; code-driven environment | *** #### 3. Infrastructure and Setup Effort | Dimension | Adiom Dsync | Dump-Restore | Native Cloud Tools | Debezium | 3rd Party ETL | Apache Spark | | ------------------------ | ------------------------------------------------- | -------------------------------------- | --------------------------------- | -------------------------------------- | ----------------------------------------- | ------------------------------------- | | **How is it Deployed?** | Kubernetes-Native (GKE/AKS/Helm) or single-binary | Local terminal tools | Managed cloud interface | Distributed Kafka Connect setup | SaaS web platform or heavy local app | Distributed big data cluster | | **Hardware Footprint** | Minimal; compute-only inside your VPC | Heavy local storage needed for files | Requires cloud migration nodes | Heavy; requires a full Kafka cluster | SaaS subscription + cloud data out costs | Heavy; large server cluster needed | | **Setup Effort** | Low; simple developer-friendly configuration | High manual operational work | Medium configuration | Extremely high setup and upkeep | Medium configuration; high monthly bills | Very high software development effort | | **Built for Migrations** | Yes; optimized for zero-downtime cutovers | No; built for database backup/recovery | Yes; but limited to target clouds | No; built for continuous event streams | No; built for analytical data warehousing | No; built for generic batch analytics | | **Enterprise Support** | Yes; dedicated support from Adiom | No; standard database manuals only | Yes; via your standard cloud tier | Community or paid Kafka vendors | Yes; standard SaaS ticketing | Depends on vendor distribution | *** ### Detailed Architectural Breakdowns #### Dsync vs. Dump-Restore Standard database dump utilities (e.g., `mongodump`, `pg_dump`) extract data to offline files that are copied over the network and re-imported into the target database. While simple, this forces a hard trade-off: either you accept hours to days of complete database downtime while the process executes, or you risk massive data drift if updates occur post-dump. * **The Dsync Advantage:** Dsync streams data directly while the source database is completely live. It combines initial reading with real-time Change Data Capture (CDC), maintaining an active-active sync state until you are ready for a sub-minute cutover window. #### Dsync vs. Native Cloud Tools (e.g., GCP Database Migration Service) Cloud provider database migration tools are generally designed to simplify landing workloads within their specific ecosystems. However, they are often restricted to highly specific version pairs and homogenous moves, and can suffer from tight performance throttling over standard cloud APIs. * **The Dsync Advantage:** Dsync avoids ecosystem lock-in, seamlessly orchestrating transfers across SQL, NoSQL, and vector backends. Furthermore, Dsync runs entirely as an isolated runtime within your VPC, bypassing external SaaS constraints to unlock max migration performance. #### Dsync vs. Debezium / Kafka Connect Debezium is a fantastic engineering framework for long-term, distributed event-driven microservices. However, deploying it solely for a database migration introduces immense operational complexity: teams must spin up, secure, and manage a complete Apache Kafka or Redpanda cluster alongside a Kafka Connect distributed runtime. * **The Dsync Advantage:** Dsync operates via a zero-dependency architecture. It's Open Source version works as standalone CLI tool. In Enterprise setups, it deploys directly onto Kubernetes cluster via a clean Helm install, executing high-speed, resumable transfers without requiring any intermediate message brokers or storage pools. #### Dsync vs. 3rd-Party ETL Platforms (e.g., Fivetran, Airbyte) Modern data integration and ELT systems are purpose-built for analytics extraction. They pull batches of business logs or operational state snapshots and drop them into analytical data warehouses (like BigQuery or Snowflake). They are not engineered to handle low-latency operational database cutovers or sync high-throughput write-heavy transactional systems. * **The Dsync Advantage:** Dsync is optimized exclusively for data mobility and system modernizations. It safely preserves index definitions, unique types, and operational performance paradigms, facilitating immediate active transactional use on the destination end. #### Dsync vs. Apache Spark Apache Spark provides a highly scalable framework for distributed big-data calculation and batch manipulation. Using it for migrations requires engineering teams to write, test, and debug massive amounts of custom data connection scripts, connection pooling profiles, and failure/resume recovery logic. * **The Dsync Advantage:** Dsync delivers the same out-of-the-box massive cluster performance without any code. Enterprise execution is simple, scalable, and completely declarative, requiring a single line configuration to safely distribute massive migration workloads across your environment. # Step By Step This is a high-level overview of the steps necessary to migrate your database with dsync while minimizing downtime. ### Prepare for the migration See [Deploy](/basics/deploy) for Dsync installation options. Make sure both source and destination databases are accessible from the host where you are planning to run the dsync process. For best performance, run dsync on a host with minimal network latency to the source and the destination. If both databases are in the same geographical region (e.g. US Central), it's best to provision a VM there for dsync. For MongoDB Atlas and Azure Cosmos DB destinations, we recommend provisioning higher throughput to avoid timeouts during the initial bulk data copy. Additionally, we recommend [enabling server-side retries](https://learn.microsoft.com/en-us/azure/cosmos-db/mongodb/prevent-rate-limiting-errors) for Azure Cosmos DB destinations, at least for the duration of the migration. ### Step 1. Start Start the dsync process and let it run until it progresses through the initial data copy and the changestream. When the reported events lag is close to 0, the destination is caught up with the source.
### Step 2. Prepare your application or service for cutover While dsync is still running, stop the writes on the source and let the lag go fully to 0. When the "change stream events" counter is no longer growing, the source and destination are in sync. **Note on deletes emulation in Cosmos DB** If you're using [Cosmos DB deletes emulation](/basics/features#deletes-emulation) for a Cosmos DB source, it's possible that there's a pending delete detection cycle. A simple restart of the dsync process will force the deletes cycle to happen. ### Step 3. Validate the data Stop the dsync process and restart it with `--verify` while **keeping all other options same**.
If dsync reports a validation error, you will need to check the log file for details. ### Step 4. Restart your application or service Your data has been fully migrated, congratulations! Now you can stop the dsync process, repoint your services to the new database and resume normal operation. # Best Practices Here you will find our technical and procedural recommendations for a successful migration, based on our collective experiences. ## General 1. Over-provision the destination for the duration of the migration process 2. Build indexes later in the CDC phase 3. Use UAT and DEV post-migration average object size to get accurate estimates for PROD data size and disk sizes 4. Monitor the process 1. CPU, Memory, Disk IO on the source and destination 2. CPU and Memory for Dsync process 3. Dsync throughput and replication lag 5. Check all the components *before* the cutover time, especially when the source and the destination been in the CDC mode for a prolonged period of time 1. Source database and services health 2. Destination database and services health 3. Dsync process(es) health and replication lag 1. When running Dsync in a single-binary mode, any errors and replication lag are reflected in the CLI progress view and the dsync log file 2. When running the Enterprise version, in addition to the web progress dashboard, it's a good idea to check the status of workflows and tasks in Temporal, as well as check for any reported errors in SigNoz log collector ## Database-specific recommendations ### Cosmos DB RU destinations 1. Enable server-side retries in Cosmos DB configuration 2. Cosmos DB RU consumes a lot of resources for batch writes (\~3-5 RU per write). When using as a destination for a large migration, make sure to provision at least 8K RUs for default settings ### MongoDB destinations: * For large (multi-TB) migrations use `w:majority` write concern (default on Atlas) and increase `replWriterThreadCount` if there's CPU and disk headroom (e.g. try 64 for M200) * Disable backups * Set oplog to a fixed size in Atlas * Use network compression in the connection string when Atlas is in a different region or availability zone - this will help save network costs ## Dsynct Enterprise: * Use a dedicated disk for Temporal database (recommended 3000 IOPS, 125 MBps) # Troubleshooting Information on various error messages and general troubleshooting # Scalable Deployment ## Overview While a single dsync already packs a powerhouse in single-binary format, it's not perfectly suitable for all use cases. For example: * Large and Very large database migrations (100GB to 100TB and beyond) * Moving or replicating data within a fleet of database instances (data mobility) For these and other use cases, we have developed a horizontally scalable deployment using [Temporal](https://temporal.io/) durable execution platform as an external Coordinator and individual dsync's as workers. This Enterprise deployment offers scalability, flexibility, and additional observability on top of Open Source dsync. ## Topology
## Components 1. [Temporal](https://temporal.io/) durable workflow execution engine * Coordinates workflows and assigns tasks to workers * Handles resumability and retries on the task level, maintains state 2. [SigNoz](https://signoz.io/) observability tool * Logs and metrics collector (OpenTelemetry) * Advanced dashboards 3. [Dsynct](https://github.com/adiom-data/public/dsynct) * Worker processes: `dsynct worker` * Runner: `dsynct run` * Initiates a new workflow (or connects to existing one) * Serves real-time web progress dashboard over HTTP ## How it works In this configuration, each worker is a standalone dsync process with its own connectors and [transformer](/basics/features#transformations) instances. A migration or replication job is structured as a Temporal workflow, including tasks for Read Planning, Initial Data Copy, and CDC. When a job is submitted to Temporal, individual tasks are allocated to dsync workers based on their suitability and availability, allowing for a high degree of parallelization. Temporal stores the job metadata and makes task execution durable. Should a worker or a task fail, Temporal will automatically reallocate the task to the next available worker. ## Distribution The Enterprise version of dsync is distributed under Commercial License that allows for a free trial period. To download and run this configuration, you can use docker containers or Kubernetes charts: . We also have versions of Kubernetes charts available for Azure, Google, and AWS services. On Azure you can use our AKS-native marketplace offering: [Azure Marketplace](/enterprise/running-dsynct/azure-marketplace). Contact us at if you're interested and want to know more. # Running Dsynct Learn about how to run the Enterprise version of Dsync # Instructions Learn how to run Enterprise Dsynct with Docker containers ## Core Components {% tabs %} {% tab title="Temporal" %} Dsynct is able to leverage an existing [Temporal](https://temporal.io/) instance, if you're already using it in your organization. Otherwise, you start the Temporal development server on a VM. It can be reused across different migrations and dsynct executions. For optimal performance, we recommend provisioning a dedicated disk for the Temporal database with the following configuration or better: 50 GB / 3000 IOPS / 125 MBps. We recommend provisioning a 4 vCPU, 8 GiB memory VM for Temporal development server. {% endtab %} {% tab title="OpenTelemetry" %} Dsynct is able to leverage an existing OpenTelemetry gRPC collector, if you're already using it in your organization. Otherwise, you start the [SigNoz](https://signoz.io/) observability tool on a VM. It can be reused across different migrations and dsynct executions. For convenience, it can be co-located with Temporal on the same VM. We recommend provisioning a 2 vCPU, 4 GiB memory VM for SigNoz. {% endtab %} {% tab title="Dsynct Worker" %} Dsync**t** is a Temporal-enabled version of Dsync. It only uses compute resources (CPU and RAM), and doesn't store the data on disk. {% hint style="info" %} Dsynct worker is the unit of scale. You can run as many as you need, and add/stop them dynamically even while the migration is running. New workers will automatically pick up tasks from the queue, and any tasks assigned to stopped workers will automatically get reallocated to those still available. {% endhint %} Each Dsynct worker instance (`dsynct worker` command) can process up to a certain number of tasks in parallel, with the level of parallelism controlled by these parameters: * `--concurrent-activities N` - max number of concurrent initial sync tasks * `--sync-transform-workers N` - parallelism for the transformer during initial sync * `--sync-writer-workers N` - parallelism for writing to the destination during initial sync * `--per-stream-workers N` - parallelism for writing to the destination during CDC Dsynct can run on a regular VM or directly as a container. We recommend provisioning at least 4 CPU and 8 GB RAM for each Dsynct worker. The level of parallelism can be adjusted based on the CPUs available, for example for a 4 CPU instance: `--concurrent-activities 4 --sync-transform-workers 4 --sync-writer-workers 8 --per-stream-workers 4` {% hint style="warning" %} If you are running multiple sets of workers with different configurations (e.g. different source/destination pairs or different transform configs), each set must use a distinct `--queue-name`. Workers sharing the same queue will receive tasks interchangeably, so mixing different configurations on the same queue will cause errors. The corresponding `dsynct run` command must also specify the same `--queue-name` to route work to the correct workers. {% endhint %} {% endtab %} {% tab title="Dsynct Runner" %} The runner (`dsynct run` command) is responsible for starting and monitoring a migration workflow. For a given flow, it ensures that it has been started on Temporal and servers a Web-based progress dashboard. The runner can run on a lightweight VM or directly as a container. We recommend provisioning 1 CPU and 2 GB RAM. {% endtab %} {% endtabs %} ## Networking
## Instructions 1. **Provision VMs or containers.** If you're using Azure Marketplace images for migrations to [Cosmos DB vCore](https://azuremarketplace.microsoft.com/en-us/marketplace/apps/adiom.adiom_dsync_vm?tab=Overview) or [Cosmos DB NoSQL](https://azuremarketplace.microsoft.com/en-us/marketplace/apps/adiom.adiom_dsync_vm_dynamo?tab=Overview), they come with all the software components preinstalled already.\ \ You can also use Kubernetes (native, AKS, or others) - see instructions [here](https://github.com/adiom-data/public/tree/main/kubernetes).\ \ We recommend provisioning at least two VMs - one for Temporal and SigNoz, and another one (or more) to run the Runner and Dsynct workers. Ubuntu on x64 is a common choice for VM. 2. \[Optional if using Marketplace images] **Install Docker**\ You can follow the official instructions [here](https://docs.docker.com/engine/install/ubuntu/#install-using-the-repository). 3. \[Optional if using Marketplace images] **Install Temporal and SigNoz** ```bash brew install temporal git clone -b main https://github.com/SigNoz/signoz.git ``` 4. \[Optional] Get most recent Docker image versions ```bash sudo docker pull markadiom/dsynct ``` 5. **Start Temporal and SigNoz**\ \&#xNAN;*When placing the Temporal database on a dedicated disk (we recommend doing that), adjust the --db-filename parameter to point at the mount point (e.g. /mnt/data.db) and ensure that the current user has write access to that folder.*\ \&#xNAN;*Note that you need to connect to SigNoz web UI (port 8080) and create an account prior to starting Dsync in order for SigNoz telemetry collector to start.* ```bash screen -dmS Temporal temporal server start-dev --db-filename data.db --ip 0.0.0.0 --dynamic-config-value limit.numPendingActivities.error=10000 --dynamic-config-value frontend.activityAPIsEnabled=true cd signoz/deploy/docker sudo docker compose up -d --remove-orphans ``` 6. **Start Dsynct worker(s)**\ For testing purposes, you can use `/dev/fakesource` as the source, and `/dev/null` as the destination. ```bash docker run -d --name dsyncworker \ -e 'OTEL_EXPORTER_OTLP_ENDPOINT=http://:4317' \ markadiom/dsynct worker \ \ \ temporal --host-port :7233 \ app --otel ``` 7. **Start the workflow** ```bash docker run --name dsyncrunner \ -p 8080:8080 \ -e 'OTEL_EXPORTER_OTLP_ENDPOINT=http://:4317' \ markadiom/dsynct run \ temporal --host-port :7233 \ app --otel --host-port 0.0.0.0:8080 ``` 8. **Monitor migration progress** 1. The Web Progress dashboard is served on the port `8080` on the host where you ran the `dsynct run` command 2. Temporal workflow can be observed by connecting to `:8233` 3. Container logs can be viewed with `docker logs -f dsyncworker` or `docker logs -f dsyncrunner` 4. You can view detailed logs and metrics in SigNoz by connecting to `:8080` You can import one of our pre-configured [dashboards](https://github.com/adiom-data/public/tree/main/kubernetes/system/signoz_dashboards) by following the instructions [here](https://signoz.io/docs/dashboards/import-dashboard/). ### Temporal Tools Dsynct can run in a `temporaltools` mode that helps make pausing and unpausing more convenient by setting the environment variable DSYNCT\_MODE=temporaltools. ``` docker run -e 'DSYNCT_MODE=temporaltools' markadiom/dsynct --help docker run -e 'DSYNCT_MODE=temporaltools' markadiom/dsynct pause --workflow-id= --dry-run docker run -e 'DSYNCT_MODE=temporaltools' markadiom/dsynct unpause --workflow-id= --dry-run ``` ## No Temporal Dsynct can also be run without temporal as a single node. This mode currently only supports a single save file to coordinate (subject to change). To run in this mode, pass set the environment variable DSYNCT\_MODE=simple and use the `sync` command. Example command that uses `resume.file` to save state: ```bash docker run -d --name dsync \ -e 'DSYNCT_MODE=simple' \ -p 8080:8080 \ -e 'OTEL_EXPORTER_OTLP_ENDPOINT=http://:4317' \ markadiom/dsynct \ --host-port=0.0.0.0:8080 \ --otel \ sync \ --save-file resume.file \ \ \ ``` The command has parameters that somewhat combine the options of the `run` and `worker` command of dsynct into one. You can run `docker run -e 'DSYNCT_MODE=simple' markadiom/dsynct sync --help` to see all available options. ### testsync The `testsync` command fetches specific documents by ID from the source, optionally transforms them, and writes them to the destination. It is useful for testing transformations or connectivity without running a full sync. It only works with connectors that support retrieving single entries via `GetByIds` (currently MongoDB and Cosmos DB NoSQL connectors). ```bash docker run -e 'DSYNCT_MODE=simple' \ markadiom/dsynct testsync \ --namespace \ --id \ ``` To test with a transformation: ```bash docker run -e 'DSYNCT_MODE=simple' \ -v "./transform.yaml:/transform.yaml" \ markadiom/dsynct testsync \ --namespace \ --id \ --transform \ dsync-transform://transform.yaml ``` | Flag | Required | Description | | --------------------- | -------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `--namespace` | Yes | The source namespace to fetch from. | | `--id` | No | Document ID (string). Can be specified multiple times for multiple documents. For composite keys, use `--id-size` to indicate how many `--id` values form a single key. | | `--jsonext-id` | No | Document ID in extended JSON format, for non-string types (e.g. `{"_id": {"$oid": "..."}}`). | | `--id-file` | No | Path to a file containing extended JSON IDs, one per line. Compatible with `sample-ids` output. | | `--id-size` | No | Number of `--id` entries that form a single composite key. Default: `1`. | | `--transform` | No | Set if a transformer is provided as the third argument after source and destination. | | `--src-data-type` | No | Source data type. Inferred if not set. | | `--dst-data-type` | No | Destination data type. Inferred if not set. | | `--namespace-mapping` | No | Namespace mapping from source to destination (e.g. `srcns:dstns`). | | `--mapping-delimiter` | No | Delimiter for namespace mappings and ID key-value pairs. Default: `:`. | At least one of `--id`, `--jsonext-id`, or `--id-file` must be provided. # DynamoDB to Cosmos DB NoSQL Steps for executing a live production migration from DynamoDB to Cosmos DB NoSQL using Dsynct. [![Adiom on Azure Marketplace](https://assetsprod.microsoft.com/mpn/en-us/ms-azure-marketplace.png)](https://azuremarketplace.microsoft.com/en/marketplace/apps/adiom.adiom_dsync_vm_dynamo) ### Prerequisites 1. **Set up Temporal and SigNoz** (or another OTEL collector). You can follow the instructions in [Running Dsynct](/enterprise/running-dsynct) 2. **Obtain DynamoDB credentials** - you can create a new dedicated user in IAM with AmazonDynamoDBFullAccess permissions, and [create security keys for it](https://docs.aws.amazon.com/sdkref/latest/guide/access-iam-users.html). 3. **Obtain Cosmos DB credentials** - URI and Primary Key ("Settings" -> "Keys" in the left menu) 4. [**Enable Streams**](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html#Streams.Enabling) **in DynamoDB** - for the tables being migrated. Use "New image". ### Worker VM setup 
## Set variables
export SIGNOZ=http://<SIGNOZ_HOST>:4317
export TEMPORAL=<TEMPORAL_HOST>:7233
export AWS_ACCESS_KEY_ID=<...>
export AWS_SECRET_ACCESS_KEY=<...>
export AWS_REGION=<...>
export COSMOS_URI=<...> #e.g. https://cosmos-nosql-west.documents.azure.com:443/
export COSMOS_KEY=<...>

## Create internal network for Docker
docker network create mynet

## Cosmos DB NoSQL Connector
docker run -d \
--network mynet --name cosmosnosqlconnector \
-e OTEL_EXPORTER_OTLP_ENDPOINT=$SIGNOZ \
markadiom/cosmosnosqlconnector 8089 $COSMOS_URI $COSMOS_KEY

## Worker
docker run -d \
--network mynet --name dsyncworker \
-e OTEL_EXPORTER_OTLP_ENDPOINT=$SIGNOZ \
-e AWS_ACCESS_KEY_ID=$AWS_ACCESS_KEY_ID \
-e AWS_SECRET_ACCESS_KEY=$AWS_SECRET_ACCESS_KEY \
-e AWS_REGION=$AWS_REGION \
markadiom/dsynct worker \
--namespace-mapping "dynamo_table:cosmos_db.container,dynamo_table1:cosmos_db.container1" \
--concurrent-activities 4 --sync-writer-workers 8 \
dynamodb \
grpc://cosmosnosqlconnector:8089 --insecure \
temporal --host-port $TEMPORAL \
app --otel
{% hint style="info" %} If data transformations are needed, you can use the embedded [YAML transformer](/enterprise/running-dsynct/data-transformations). {% endhint %} Use `docker logs -f cosmosnosqlconnector` or `docker logs -f dsyncworker` to follow container logs. ### Running the workflow ```bash ## Runner docker run \ --name dsyncrunner \ -p 8080:8080 \ -e OTEL_EXPORTER_OTLP_ENDPOINT=$SIGNOZ \ markadiom/dsynct run \ --namespace "dynamo_table" --namespace "dynamo_table1" \ temporal --host-port $TEMPORAL \ app --otel --host-port 0.0.0.0:8080 ``` # Cosmos DB NoSQL to MongoDB Steps for executing a live production migration from Cosmos DB NoSQL to MongoDB using Dsynct. ### Prerequisites 1. **Set up Temporal and SigNoz** (or another OTEL collector). You can follow the instructions in [Running Dsynct](/enterprise/running-dsynct) 2. **Obtain Cosmos DB credentials** - URI and Primary Key ("Settings" -> "Keys" in the Azure Portal) 3. **Obtain MongoDB connection string** - for the destination cluster 4. **Enable** [**"All Versions and Deletes"**](https://learn.microsoft.com/en-us/azure/cosmos-db/change-feed-modes?tabs=latest-version#all-versions-and-deletes-change-feed-mode-preview) for your Cosmos DB container ("Settings" -> "Features" in the Azure Portal) * If you can't turn on the feature, you can run the Cosmos DB NoSQL connector with the `COSMOS_DISABLE_ALL_VERSIONS_AND_DELETES=true` environment variable. In that case Dsync will not be able to replicate delete events from Cosmos DB. ### Data type and ID considerations Cosmos DB NoSQL uses **JSON** while MongoDB uses **BSON**, so a [data transformation](/enterprise/running-dsynct/data-transformations) is required. The Cosmos DB NoSQL [ID format](/reference/connectors/cosmos-db-nosql#id-format) is composed of the shard key followed by the `id` field. MongoDB uses a single `_id` field. A transform config must map between these ID formats. See [Transform Data Types](/enterprise/running-dsynct/data-types) for full details on JSON to BSON mappings. #### Simple case: shard key is `/id` When the shard key is `/id`, the Cosmos DB ID contains only the `id` field. The transform maps `id` to `_id`: ```yaml # transform.yaml defaultmapping: default mappings: - namespace: default delete: ["id"] add: ["_id"] mapid: id cel: _id: id ``` #### With a shard key prefix When the shard key is a separate field (e.g. `/region`), the Cosmos DB ID is multi-part (e.g. `["us-east", "123"]`). You need to use `idkeys` to declare the source ID fields and collapse them into a single `_id`: ```yaml # transform.yaml defaultmapping: default idlist: true mappings: - namespace: default idkeys: ["region", "id"] delete: ["region", "id"] add: ["_id"] mapid: id[1] cel: _id: id[1] ``` {% hint style="info" %} Adjust `idkeys` and the `cel` expressions to match your Cosmos DB container's shard key configuration. See the [multi-part ID examples](/enterprise/running-dsynct/data-types#multi-part-id-examples) for more patterns. {% endhint %} ### Worker VM setup ```bash ## Set variables export SIGNOZ=http://:4317 export TEMPORAL=:7233 export COSMOS_URI=<...> #e.g. https://cosmos-nosql-west.documents.azure.com:443/ export COSMOS_KEY=<...> export MONGODB_URI=<...> #e.g. mongodb+srv://user:pass@cluster.mongodb.net ## Create internal network for Docker docker network create mynet ## Cosmos DB NoSQL Connector (source) docker run -d \ --network mynet --name cosmosnosqlconnector \ -e OTEL_EXPORTER_OTLP_ENDPOINT=$SIGNOZ \ markadiom/cosmosnosqlconnector 8089 $COSMOS_URI $COSMOS_KEY ## Worker docker run -d \ --network mynet --name dsyncworker \ -e OTEL_EXPORTER_OTLP_ENDPOINT=$SIGNOZ \ -v "./transform.yaml:/transform.yaml" \ markadiom/dsynct worker \ --namespace-mapping "cosmos_db.container:mongo_db.collection" \ --concurrent-activities 4 --sync-writer-workers 8 \ --transform \ grpc://cosmosnosqlconnector:8089 --insecure \ $MONGODB_URI \ dsync-transform:///transform.yaml \ temporal --host-port $TEMPORAL \ app --otel ``` {% hint style="info" %} If no ID or data type transformation is needed, you can omit the `--transform` flag, the `dsync-transform://` argument, and the `-v` volume mount. {% endhint %} Use `docker logs -f cosmosnosqlconnector` or `docker logs -f dsyncworker` to follow container logs. ### Running the workflow ```bash ## Runner docker run \ --name dsyncrunner \ -p 8080:8080 \ -e OTEL_EXPORTER_OTLP_ENDPOINT=$SIGNOZ \ markadiom/dsynct run \ --namespace "cosmos_db.container" \ temporal --host-port $TEMPORAL \ app --otel --host-port 0.0.0.0:8080 ``` # Data Transformations ## Overview Enterprise Dsync features a prebuilt YAML-based transformer that allows to add, remove and modify data elements using mappings and Common Expression Language ([CEL](https://github.com/google/cel-spec)). Transformations are applied on-the-fly during both initial sync and CDC. The transformer itself can be run as a standalone process that Dsynct connects to over gRPC. The detailed instructions can be found in our [public repository.](https://github.com/adiom-data/public/tree/main/dsync-transform) For convenience, Dsynct workers can run the transformer as an embedded process by providing the `--transform` option along with the path to the config file as the third argument: ```bash dsynct worker --transform $SOURCE $DESTINATION dsync-transform://transform.yaml ``` When using Docker to run Dsynct, the transformer config file needs to be mounted to the container. For example: ``` docker run \ -v "./transform.yaml:/transform.yaml" \ markadiom/dsynct worker \ --transform \ $SOURCE $DESTINATION dsync-transform://transform.yaml ``` ## Writing Config Files Dsync Transform runs off a YAML configuration file where the mappings are specified. Each source document is converted into an internal format, subjected to the mappings, and then converted back to the output document type. Duplicate mappings are allowed and will fan out. Use the `filter` feature to avoid fanout if necessary. Note that mapping IDs should also require mapping the ID keys. ID mappings only have access to the original ID so that they can be applied to deletes which do not have access to the full data. If you need to convert ID types between systems (e.g., string to BSON ObjectID), see the [Transform Data Types](/enterprise/running-dsynct/data-types) page for detailed guidance and examples. ### Example Config File ```yaml mappings: - namespace: srcnamespace mapnamespace: dstnamespace map: should_be_int32: int32 cel: name: self + "!" newfield: '"abcd"' should_be_int32: self + 5 add: ["newfield"] delete: ["existingfield"] ``` Each mapping must specify the source `namespace`. If the destination namespace is different, use `mapnamespace`. The key fields work as follows: * **`cel`** -- Specify a CEL expression for mapping each field. The variable `self` refers to the current value of that field. Note that CEL only supports a limited set of types (e.g., 64-bit integers but not 32-bit integers). * **`map`** -- Apply a special type mapping *after* the `cel` expression. Use this when you need a type that CEL cannot represent directly, such as `int32` for a 32-bit integer. * **`add`** -- List fields that should be created in the output even if they were not present in the source document. * **`delete`** -- List fields that should be removed from the output. ### Configuration Reference #### Top-Level Options | Option | Type | Default | Description | | ----------------- | ----------------- | ------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------- | | `wild` | string | `*` | When specifying a path, this matches anything. | | `delimiter` | string | `.` | When specifying a path, this is the delimiter. | | `env` | map\[string, any] | | Variables available under the `env` variable in CEL expression mappings. | | `unwrapbson` | boolean | `false` | If true, will automatically convert various BSON types to a more native type (e.g., ObjectIDs to strings). | | `filtererrors` | boolean | `false` | If true, will not fail on errors during conversion and instead skip and log a warning. Errors encountered when retrieving the original ID will still error. | | `defaultmapping` | string | | Name (namespace) of the mapping from `mappings` to use as a fallback. | | `namespacemapper` | CEL string | | Default expression to automatically map all namespaces. Has `env` and `self` (namespace) available. | | `idlist` | boolean | `false` | If true, the `id` variable will always be a list. When false, the `id` variable contains the first id value if it is the only id value. | | `mappings` | list\[mapping] | | List of mapping definitions (see below). | #### Mapping Options Each entry in `mappings` supports the following fields: | Option | Type | Description | | -------------- | ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------ | | `namespace` | string | Namespace this mapping applies to. | | `mapnamespace` | string | New namespace name for the output. | | `mapid` | CEL string | Expression to map the `id` for updates. Only has the original `id` field and `env` available. | | `filter` | CEL string | Expression that returns a boolean; if true, the document will be retained. Only has the original `id` field and `env` available. | | `idkeys` | list\[string] | Describes the original names of each part of the id. | | `finalidkeys` | list\[string] | Describes the names of each part of the id after the mapping. | | `add` | list\[string] | Paths that will be added in the mapping if the parent exists. | | `delete` | list\[string] | Paths that will be deleted if they exist. | | `cel` | map\[string, CEL string] | For each defined path, specify a CEL expression to perform a mapping. Has `env`, `id`, `doc`, `parent`, and `self` available as variables. | | `map` | map\[string, string] | For each defined path, specify a special mapping function to apply. Applies after `cel`. | | `self` | CEL string | An expression that serves as a mapping for the whole document. | ### Available Mappings For use with the `map` configuration or inside a `cel` configuration expression. In certain cases, it is advisable to use `map` to force a type that CEL cannot represent directly. #### Type Conversions | Mapping | Description | | ------------- | ------------------------------------------------ | | `int32` | Converts to an int32. Should use in `map` only. | | `float` | Converts to a float32. Should use in `map` only. | | `json_number` | Converts to a JSON Number. | | `json_decode` | Decodes a JSON string or bytes into an object. | | `json_encode` | Encodes an object as a JSON string. | #### BSON Conversions | Mapping | Description | | ------------------------ | --------------------------------------------------------------- | | `bson_decimal128` | Converts a string to a BSON Decimal128. | | `bson_decimal128_string` | Converts a BSON Decimal128 to a string. | | `bson_object_id` | Converts a string to a BSON ObjectID. Should use in `map` only. | | `bson_uuid` | Converts a UUID string to a BSON UUID. | | `bson_object_id_string` | Converts BSON ObjectID to a string. | | `bson_uuid_string` | Converts BSON UUID to a string. | #### Hash Functions | Mapping | Description | | -------- | --------------------------------------------------------------- | | `md5` | Applies the MD5 hash to a string or bytes, returning bytes. | | `sha1` | Applies the SHA-1 hash to a string or bytes, returning bytes. | | `sha256` | Applies the SHA-256 hash to a string or bytes, returning bytes. | #### Byte Mappings | Mapping | Description | | --------------- | ------------------------------------------------------------------------------- | | `be_to_int32` | Converts bytes to an int assuming big endian format. Use in `map` to get int32. | | `be_to_int64` | Converts bytes to an int64 assuming big endian format. | | `to_be_int32` | Converts data into bytes representing an int32 in big endian format. | | `to_be_int64` | Converts data into bytes representing an int64 in big endian format. | | `reverse_bytes` | Reverses a byte array. | ### Available Functions All the available mappings above are usable as unary functions in CEL expressions. The following additional functions are also available: | Function | Description | | ---------------------------- | ------------------------------------------------------------------------------------- | | `now_millis()` | Current time in milliseconds. | | `now_nanos()` | Current time in nanoseconds (resolution may be limited by your machine). | | `uuid_v4_bytes()` | Generate a random UUID as bytes. | | `uuid_v4_string()` | Generate a random UUID as a string. | | `uuid_v3_bytes(uuid, name)` | Generate a deterministic UUID based on a UUID namespace and name as bytes (MD5). | | `uuid_v3_string(uuid, name)` | Generate a deterministic UUID based on a UUID namespace and name as a string (MD5). | | `uuid_v5_bytes(uuid, name)` | Generate a deterministic UUID based on a UUID namespace and name as bytes (SHA-1). | | `uuid_v5_string(uuid, name)` | Generate a deterministic UUID based on a UUID namespace and name as a string (SHA-1). | #### Fake Data Generation Generate deterministic fake data seeded by the input value. The same seed always produces the same output, making these suitable for consistent data anonymization. Numeric types are used directly as seeds; other types (strings, etc.) are hashed to derive a seed. | Function | Description | | ----------------------- | ------------------------------- | | `fake_name(seed)` | Generate a fake full name. | | `fake_first_name(seed)` | Generate a fake first name. | | `fake_last_name(seed)` | Generate a fake last name. | | `fake_email(seed)` | Generate a fake email address. | | `fake_phone(seed)` | Generate a fake phone number. | | `fake_address(seed)` | Generate a fake street address. | | `fake_city(seed)` | Generate a fake city name. | | `fake_state(seed)` | Generate a fake state name. | | `fake_zip(seed)` | Generate a fake postal code. | | `fake_country(seed)` | Generate a fake country name. | | `fake_company(seed)` | Generate a fake company name. | | `fake_username(seed)` | Generate a fake username. | | `fake_ipv4(seed)` | Generate a fake IPv4 address. | | `fake_sentence(seed)` | Generate a fake sentence. | | `fake_word(seed)` | Generate a fake word. | | `fake_url(seed)` | Generate a fake URL. | For the latest details, consult the README in our [public repository.](https://github.com/adiom-data/public/tree/main/dsync-transform) ## Transform Studio In order to facilitate testing out transformations, you can run the dsync web-api. ``` docker run -e DSYNCT_MODE=simple -p 8080:8080 markadiom/dsynct --host-port 0.0.0.0:8080 web-api --simple-only ``` Then open up your browser to the specified address and navigate to `Tools > Transform Studio`. The interface will allow you to test out a transform config and JSON/BSON documents. For BSON documents, use extended JSON encoding. The update keys should be specified in extended JSON encoding as well. Example extended json: ``` {"$oid": "<24-character string>"} # Mongo Object ID {"$date": ""} # Date ``` # Transform Data Types ## Data Format During a sync, data can appear in three different formats depending on the operation: 1. **Initial Sync** -- A single binary blob representing the full document. 2. **Updates** -- A list representing the ID fields, plus a binary blob of the updated data. 3. **Deletes** -- A list representing only the ID fields. A correct mapping must handle all three cases. This means you may need to provide a mapping for both the data (to cover case 1) and the ID (to cover cases 2 and 3). ## Data Types Connectors may support different data types. Currently we support **JSON** and **BSON**. When transferring data between connectors that use different types, you need to provide configuration to map between them. ### ID Mapping The ID is particularly important to transform because the data type and/or field names may differ between source and destination. **Defaults:** * **JSON**: single field called `id` * **BSON**: single field called `_id` IDs can also be composed of multiple fields. Use the `idkeys` (source) and `finalidkeys` (destination) properties in the config if the ID format does not match the defaults on either side. **General approach:** * Specify new fields under the `add` property. * Specify old fields to remove under the `delete` property. * Define a `mapid` expression so that update IDs can be set correctly. * Add a `cel` expression for each new ID field showing how it is populated from the data. #### The `id` Variable in CEL In CEL expressions, `id` is a built-in variable representing the document's ID. * If the ID has **one field**, `id` is the value of that field directly. * If the ID has **multiple fields**, `id` is a list of values. ```cel # Given id: {a: 1, b: 2} id[1] # returns 2 # Given id: {b: 2} id # returns 2 ``` You can set `idlist: true` at the top level of the config to force `id` to always be a list, even when the ID contains only one field. #### JSON to BSON Examples Rename a string `id` (`"123"`) to a string `_id` (`"123"`): ```yaml defaultmapping: default mappings: - namespace: default delete: ["id"] add: ["_id"] mapid: id cel: _id: id ``` Map a string `id` (`"123"`) to an ObjectID `_id` (`{"$oid": "202cb962ac59075b964b0715"}`): ```yaml defaultmapping: default mappings: - namespace: default delete: ["id"] add: ["_id"] mapid: id cel: _id: id map: _id: bson_object_id ``` #### BSON to JSON Examples Rename a string `_id` (`"123"`) to a string `id` (`"123"`): ```yaml defaultmapping: default mappings: - namespace: default delete: ["_id"] add: ["id"] mapid: id cel: id: id ``` Convert an ObjectID `_id` (`{"$oid": "202cb962ac59075b964b0715"}`) to a string `id` (`"202cb962ac59075b964b0715"`): ```yaml defaultmapping: default mappings: - namespace: default delete: ["_id"] add: ["id"] mapid: string(id) cel: id: string(id) ``` #### Multi-Part ID Examples When the source ID is composed of multiple fields, `id` becomes a list. Use `idkeys` to declare the source ID fields and `finalidkeys` for the destination. Individual parts are accessed with `id[0]`, `id[1]`, etc. Map a two-part JSON ID (`region` and `user_id`) to BSON, renaming them to `_region` and `_user_id`: ```yaml defaultmapping: default mappings: - namespace: default idkeys: ["region", "user_id"] finalidkeys: ["_region", "_user_id"] delete: ["region", "user_id"] add: ["_region", "_user_id"] mapid: id cel: _region: id[0] _user_id: id[1] ``` Collapse a two-part JSON ID (`tenant` and `record_id`) into a single BSON `_id` string by concatenating them: ```yaml defaultmapping: default mappings: - namespace: default idkeys: ["tenant", "record_id"] delete: ["tenant", "record_id"] add: ["_id"] mapid: string(id[0]) + ":" + string(id[1]) cel: _id: string(id[0]) + ":" + string(id[1]) ``` Expand a single BSON `_id` back into a two-part JSON ID by splitting on a delimiter: ```yaml defaultmapping: default mappings: - namespace: default finalidkeys: ["tenant", "record_id"] delete: ["_id"] add: ["tenant", "record_id"] mapid: '[string(id).split(":")[0], string(id).split(":")[1]]' cel: tenant: string(id).split(":")[0] record_id: string(id).split(":")[1] ``` # CEL Reference This is a reference for Common Expression Language (CEL) methods, functions, and macros available in Dsynct transformations. ## Methods ### bytes | Method | Return Type | | -------------- | ----------- | | `bytes.size()` | `int` | ### google.protobuf.Duration | Method | Return Type | | -------------------------------------------- | ----------- | | `google.protobuf.Duration.getHours()` | `int` | | `google.protobuf.Duration.getMilliseconds()` | `int` | | `google.protobuf.Duration.getMinutes()` | `int` | | `google.protobuf.Duration.getSeconds()` | `int` | ### google.protobuf.Timestamp | Method | Return Type | | --------------------------------------------------- | ----------- | | `google.protobuf.Timestamp.getDate()` | `int` | | `google.protobuf.Timestamp.getDate(string)` | `int` | | `google.protobuf.Timestamp.getDayOfMonth()` | `int` | | `google.protobuf.Timestamp.getDayOfMonth(string)` | `int` | | `google.protobuf.Timestamp.getDayOfWeek()` | `int` | | `google.protobuf.Timestamp.getDayOfWeek(string)` | `int` | | `google.protobuf.Timestamp.getDayOfYear()` | `int` | | `google.protobuf.Timestamp.getDayOfYear(string)` | `int` | | `google.protobuf.Timestamp.getFullYear()` | `int` | | `google.protobuf.Timestamp.getFullYear(string)` | `int` | | `google.protobuf.Timestamp.getHours()` | `int` | | `google.protobuf.Timestamp.getHours(string)` | `int` | | `google.protobuf.Timestamp.getMilliseconds()` | `int` | | `google.protobuf.Timestamp.getMilliseconds(string)` | `int` | | `google.protobuf.Timestamp.getMinutes()` | `int` | | `google.protobuf.Timestamp.getMinutes(string)` | `int` | | `google.protobuf.Timestamp.getMonth()` | `int` | | `google.protobuf.Timestamp.getMonth(string)` | `int` | | `google.protobuf.Timestamp.getSeconds()` | `int` | | `google.protobuf.Timestamp.getSeconds(string)` | `int` | ### list | Method | Return Type | | ---------------------------------------- | --------------------------------- | | `list.size()` | `int` | | `list.distinct()` | `list` | | `list.reverse()` | `list` | | `list.slice(int, int)` | `list` | | `list.first()` | `optional_type` | | `list.last()` | `optional_type` | | `list.sort()` | `list` | | `list.sort()` | `list` | | `list.sort()` | `list` | | `list.flatten(int)` | `list` | | `list.sort()` | `list` | | `list.sort()` | `list` | | `list.sort()` | `list` | | `list>.flatten()` | `list` | | `list>.unwrapOpt()` | `list` | | `list.join()` | `string` | | `list.join(string)` | `string` | | `list.sort()` | `list` | | `list.sort()` | `list` | ### map | Method | Return Type | | ------------------ | ----------- | | `map.size()` | `int` | ### optional\_type | Method | Return Type | | --------------------------------------- | ------------------ | | `optional_type.hasValue()` | `bool` | | `optional_type.or(optional_type)` | `optional_type` | | `optional_type.orValue(V)` | `V` | | `optional_type.value()` | `V` | ### string | Method | Return Type | | ------------------------------------- | -------------- | | `string.charAt(int)` | `string` | | `string.contains(string)` | `bool` | | `string.endsWith(string)` | `bool` | | `string.format(list)` | `string` | | `string.indexOf(string)` | `int` | | `string.indexOf(string, int)` | `int` | | `string.lastIndexOf(string)` | `int` | | `string.lastIndexOf(string, int)` | `int` | | `string.lowerAscii()` | `string` | | `string.matches(string)` | `bool` | | `string.replace(string, string)` | `string` | | `string.replace(string, string, int)` | `string` | | `string.reverse()` | `string` | | `string.size()` | `int` | | `string.split(string)` | `list` | | `string.split(string, int)` | `list` | | `string.startsWith(string)` | `bool` | | `string.substring(int)` | `string` | | `string.substring(int, int)` | `string` | | `string.trim()` | `string` | | `string.upperAscii()` | `string` | ## Functions ### Base64 | Function | Return Type | | ----------------------- | ----------- | | `base64.decode(string)` | `bytes` | | `base64.encode(bytes)` | `string` | ### Byte Order | Function | Return Type | | -------------------- | ----------- | | `be_to_int32(dyn)` | `dyn` | | `be_to_int64(dyn)` | `dyn` | | `reverse_bytes(dyn)` | `bytes` | | `to_be_int32(dyn)` | `bytes` | | `to_be_int64(dyn)` | `bytes` | ### BSON | Function | Return Type | | ----------------------------- | ----------- | | `bson_decimal128(dyn)` | `dyn` | | `bson_decimal128_string(dyn)` | `string` | | `bson_object_id_string(dyn)` | `string` | | `bson_uuid(dyn)` | `dyn` | | `bson_uuid_string(dyn)` | `string` | ### Type Conversion | Function | Return Type | | ----------------------------------- | ----------- | | `bool(bool)` | `bool` | | `bool(string)` | `bool` | | `bytes(bytes)` | `bytes` | | `bytes(string)` | `bytes` | | `double(double)` | `double` | | `double(int)` | `double` | | `double(string)` | `double` | | `double(uint)` | `double` | | `dyn(A)` | `dyn` | | `int(double)` | `int` | | `int(google.protobuf.Duration)` | `int` | | `int(google.protobuf.Timestamp)` | `int` | | `int(int)` | `int` | | `int(string)` | `int` | | `int(uint)` | `int` | | `string(bool)` | `string` | | `string(bytes)` | `string` | | `string(double)` | `string` | | `string(google.protobuf.Duration)` | `string` | | `string(google.protobuf.Timestamp)` | `string` | | `string(int)` | `string` | | `string(string)` | `string` | | `string(uint)` | `string` | | `type(A)` | `type` | | `uint(double)` | `uint` | | `uint(int)` | `uint` | | `uint(string)` | `uint` | | `uint(uint)` | `uint` | ### Duration and Timestamp | Function | Return Type | | -------------------------------------- | --------------------------- | | `duration(google.protobuf.Duration)` | `google.protobuf.Duration` | | `duration(int)` | `google.protobuf.Duration` | | `duration(string)` | `google.protobuf.Duration` | | `timestamp(google.protobuf.Timestamp)` | `google.protobuf.Timestamp` | | `timestamp(int)` | `google.protobuf.Timestamp` | | `timestamp(string)` | `google.protobuf.Timestamp` | | `now_millis()` | `int` | | `now_nanos()` | `int` | ### JSON | Function | Return Type | | ------------------ | ----------- | | `json_decode(dyn)` | `dyn` | | `json_encode(dyn)` | `dyn` | | `json_number(dyn)` | `dyn` | ### Collections | Function | Return Type | | ----------------------------------- | ----------- | | `in(A, list)` | `bool` | | `in(A, map)` | `bool` | | `lists.range(int)` | `list` | | `sets.contains(list, list)` | `bool` | | `sets.equivalent(list, list)` | `bool` | | `sets.intersects(list, list)` | `bool` | | `size(bytes)` | `int` | | `size(list)` | `int` | | `size(map)` | `int` | | `size(string)` | `int` | ### Math | Function | Return Type | | ------------------------------- | ----------- | | `math.abs(double)` | `double` | | `math.abs(int)` | `int` | | `math.abs(uint)` | `uint` | | `math.bitAnd(int, int)` | `int` | | `math.bitAnd(uint, uint)` | `uint` | | `math.bitNot(int)` | `int` | | `math.bitNot(uint)` | `uint` | | `math.bitOr(int, int)` | `int` | | `math.bitOr(uint, uint)` | `uint` | | `math.bitShiftLeft(int, int)` | `int` | | `math.bitShiftLeft(uint, int)` | `uint` | | `math.bitShiftRight(int, int)` | `int` | | `math.bitShiftRight(uint, int)` | `uint` | | `math.bitXor(int, int)` | `int` | | `math.bitXor(uint, uint)` | `uint` | | `math.ceil(double)` | `double` | | `math.floor(double)` | `double` | | `math.isFinite(double)` | `bool` | | `math.isInf(double)` | `bool` | | `math.isNaN(double)` | `bool` | | `math.round(double)` | `double` | | `math.sign(double)` | `double` | | `math.sign(int)` | `int` | | `math.sign(uint)` | `uint` | | `math.sqrt(double)` | `double` | | `math.sqrt(int)` | `double` | | `math.sqrt(uint)` | `double` | | `math.trunc(double)` | `double` | ### Hashing | Function | Return Type | | ------------- | ----------- | | `md5(dyn)` | `bytes` | | `sha1(dyn)` | `bytes` | | `sha256(dyn)` | `bytes` | ### Optional | Function | Return Type | | ----------------------------------------- | ------------------ | | `optional.none()` | `optional_type` | | `optional.of(V)` | `optional_type` | | `optional.ofNonZeroValue(V)` | `optional_type` | | `optional.unwrap(list>)` | `list` | ### Regex | Function | Return Type | | -------------------------------------------- | ----------------------- | | `matches(string, string)` | `bool` | | `regex.extract(string, string)` | `optional_type` | | `regex.extractAll(string, string)` | `list` | | `regex.replace(string, string, string)` | `string` | | `regex.replace(string, string, string, int)` | `string` | ### Strings | Function | Return Type | | ----------------------- | ----------- | | `strings.quote(string)` | `string` | ### UUID | Function | Return Type | | -------------------------- | ----------- | | `uuid_v3_bytes(dyn, dyn)` | `bytes` | | `uuid_v3_string(dyn, dyn)` | `string` | | `uuid_v4_bytes()` | `bytes` | | `uuid_v4_string()` | `string` | | `uuid_v5_bytes(dyn, dyn)` | `bytes` | | `uuid_v5_string(dyn, dyn)` | `string` | ## Macros Macros are special constructs that are expanded at parse time and provide control flow and comprehension capabilities that cannot be expressed as regular functions. ### has Tests whether a field is available on a message or key exists in a map. **Signature:** `has(e.f)` -> `bool` **Examples:** ```cel has(user.address) // true if 'address' field exists in the 'user' message has(m.key_name) // true if map 'm' has a key named 'key_name' has(order.items) // false if 'items' field is not set ``` ### all Tests whether all elements in a list or all keys in a map satisfy the given predicate. Behaves consistently with logical AND, including short-circuit evaluation and error absorption. **Signatures:** * `list(A).all(x, predicate)` -> `bool` * `map(A, B).all(x, predicate)` -> `bool` **Examples:** ```cel [1, 2, 3].all(x, x > 0) // true [1, 2, 0].all(x, x > 0) // false ['apple', 'banana'].all(fruit, fruit.size() > 3) // true {'a': 1, 'b': 2}.all(key, key != 'b') // false ``` ### exists Tests whether any element in a list or any key in a map satisfies the predicate. Behaves consistently with logical OR, including short-circuit evaluation and error absorption. **Signatures:** * `list(A).exists(x, predicate)` -> `bool` * `map(A, B).exists(x, predicate)` -> `bool` **Examples:** ```cel [1, 2, 3].exists(i, i % 2 != 0) // true [].exists(i, i > 0) // false [0, -1, 5].exists(num, num < 0) // true {'x': 'foo', 'y': 'bar'}.exists(key, key == 'x') // true ``` ### exists\_one Tests whether exactly one element in a list or key in a map satisfies the predicate. Does not short-circuit to ensure exactly one match. **Signatures:** * `list(A).exists_one(x, predicate)` -> `bool` * `map(A, B).exists_one(x, predicate)` -> `bool` **Examples:** ```cel [1, 2, 2].exists_one(i, i < 2) // true (only 1 is < 2) [1, 2, 3, 4].exists_one(num, num % 2 == 0) // false (2 and 4 are even) {'a': 'hello', 'aa': 'world'}.exists_one(k, k.startsWith('a')) // false (both match) ``` ### map Transforms each element in a list or each key in a map by applying an expression. The three-argument form transforms all elements; the four-argument form only transforms elements matching a filter predicate. **Signatures:** * `list(A).map(x, transform)` -> `list(T)` * `list(A).map(x, filter, transform)` -> `list(T)` * `map(A, B).map(x, transform)` -> `list(T)` * `map(A, B).map(x, filter, transform)` -> `list(T)` **Examples:** ```cel [1, 2, 3].map(x, x * 2) // [2, 4, 6] [5, 10, 15].map(x, x / 5) // [1, 2, 3] ['apple', 'banana'].map(fruit, fruit.upperAscii()) // ['APPLE', 'BANANA'] [1, 2, 3, 4].map(num, num % 2 == 0, num * 2) // [4, 8] (only even numbers) ``` ### filter Returns a list containing only the elements from the input list or keys from a map that satisfy the predicate. **Signatures:** * `list(A).filter(x, predicate)` -> `list(A)` * `map(A, B).filter(x, predicate)` -> `list(A)` **Examples:** ```cel [1, 2, 3].filter(x, x > 1) // [2, 3] ['cat', 'dog', 'bird', 'fish'].filter(pet, pet.size() == 3) // ['cat', 'dog'] {'a': 1, 'b': 2, 'c': 3}.filter(key, key != 'b') // ['a', 'c'] ``` ### cel.bind Binds an identifier to an initialization expression for use in a subsequent result expression. Useful for avoiding repeated computation of complex sub-expressions. **Signature:** `cel.bind(varName, initExpr, resultExpr)` **Examples:** ```cel cel.bind(x, self.field1 + self.field2, x * x) // computes sum once, then squares it cel.bind(name, user.firstName + ' ' + user.lastName, name.size() > 0 && name != 'Anonymous') ``` ### optMap Applies a transformation to the value inside an optional if it has a value, otherwise returns `optional.none()`. **Signature:** `optional(V).optMap(x, transform)` -> `optional(T)` **Examples:** ```cel optional.of(5).optMap(x, x * 2) // optional.of(10) optional.none().optMap(x, x * 2) // optional.none() ``` ### optFlatMap Like `optMap`, but the transformation must return an optional. Useful for chaining optional operations without nested optionals. **Signature:** `optional(V).optFlatMap(x, transform)` -> `optional(T)` **Examples:** ```cel optional.of(5).optFlatMap(x, optional.of(x * 2)) // optional.of(10) optional.of(5).optFlatMap(x, optional.none()) // optional.none() ``` ### math.least Returns the minimum value from a variable number of arguments. All arguments must be numeric (int, uint, or double). **Signature:** `math.least(arg1, arg2, ...)` -> `double|int|uint` **Examples:** ```cel math.least(1, 2, 3) // 1 math.least(3.14, 2.71, 1.0) // 1.0 math.least(5u, 3u, 8u) // 3u ``` ### math.greatest Returns the maximum value from a variable number of arguments. All arguments must be numeric (int, uint, or double). **Signature:** `math.greatest(arg1, arg2, ...)` -> `double|int|uint` **Examples:** ```cel math.greatest(1, 2, 3) // 3 math.greatest(3.14, 2.71, 1.0) // 3.14 math.greatest(5u, 3u, 8u) // 8u ``` ### sortBy Sorts a list by a key expression. Each element is transformed by the key expression, and the list is sorted by those keys. **Signature:** `list(T).sortBy(x, keyExpr)` -> `list(T)` **Examples:** ```cel [{'name': 'bob', 'age': 30}, {'name': 'alice', 'age': 25}].sortBy(p, p.age) // [{'name': 'alice', 'age': 25}, {'name': 'bob', 'age': 30}] ['banana', 'apple', 'cherry'].sortBy(s, s.size()) // ['apple', 'banana', 'cherry'] (sorted by string length) ``` # Verification Dsynct provides several commands for verifying that data matches between source and destination. All verification commands run in simple mode (`DSYNCT_MODE=simple`). These features are currently experimental. ## verify The `verify` command performs a full verification by reading all data from both the source and destination connectors in parallel and comparing them. It supports both initial sync verification and ongoing change stream verification. ```bash docker run -e 'DSYNCT_MODE=simple' \ markadiom/dsynct verify \ --namespace \ ``` With transformations (source-side and/or destination-side): ```bash docker run -e 'DSYNCT_MODE=simple' \ -v "./transform.yaml:/transform.yaml" \ markadiom/dsynct verify \ --namespace \ --src-transform \ dsync-transform://transform.yaml ``` | Flag | Required | Description | | ----------------------- | -------- | --------------------------------------------------------------------------------------------------------------- | | `--namespace` | No | Source namespace(s). Can be specified multiple times. | | `--dst-namespace` | No | Destination namespace(s). Defaults to source namespaces (or mapped namespaces). | | `--namespace-mapping` | No | Namespace mapping from source to destination. | | `--parallelism` | No | Number of parallel workers. Default: `1`. | | `--src-transform` | No | Set if a source-side transformer is provided after the two connectors. | | `--dst-transform` | No | Set if a destination-side transformer is provided (after the source transformer if present). | | `--src-data-type` | No | Source data type. Inferred if not set. | | `--dst-data-type` | No | Destination data type. Inferred if not set. | | `--transform-data-type` | No | Intermediate comparison data type. Default: `DATA_TYPE_MONGO_BSON`. | | `--skip-initial-sync` | No | Skip initial sync verification. | | `--skip-change-stream` | No | Skip change stream verification. | | `--latency` | No | Only compare documents that have not been updated for this duration during change stream mode. Default: `20s`. | | `--report-interval` | No | How often to print progress reports. Default: `1s`. | | `--report-limit` | No | Maximum number of mismatches to report per interval. Default: `5`. | | `--report-all` | No | Report all mismatches instead of limiting. | | `--projection` | No | JSON describing which fields to include in comparisons (e.g. `{"field": {"inner_field": true}}`). | | `--id-key` | No | Field name(s) that make up the document ID. Can be specified multiple times for composite keys. Default: `_id`. | | `--partition` | No | Partition number (0-indexed) for distributed verification. Default: `0`. | | `--total-partitions` | No | Total number of partitions for distributed verification. Default: `1`. | | `--mapping-delimiter` | No | Delimiter for namespace mappings. Default: `:`. | ## sample-ids The `sample-ids` command samples document IDs from a source namespace using reservoir sampling. The output can be written to a file for later use with `verify-ids --id-file` or `testsync --id-file`. ```bash docker run -e 'DSYNCT_MODE=simple' \ markadiom/dsynct sample-ids \ --namespace \ --count 100 \ --output ids.jsonl \ ``` To sample IDs after a transformation (so the IDs reflect the transformed data): ```bash docker run -e 'DSYNCT_MODE=simple' \ -v "./transform.yaml:/transform.yaml" \ markadiom/dsynct sample-ids \ --namespace \ --count 100 \ --output ids.jsonl \ --transform \ dsync-transform://transform.yaml ``` | Flag | Required | Description | | -------------------------- | -------- | --------------------------------------------------------------------------------------------------------------------------------------- | | `--namespace` | Yes | The source namespace to sample from. | | `--count` | No | Number of IDs to sample. Default: `100`. | | `--output` | No | Output file path. Defaults to stdout. | | `--max-iter-per-partition` | No | Maximum number of ListData iterations per partition. `0` for unlimited. | | `--transform` | No | Set if a transformer is provided after the source connector. | | `--src-data-type` | No | Source data type. Inferred if not set. | | `--dst-data-type` | No | Data type after transform. Inferred if not set. | | `--id-key` | No | Field name(s) that make up the document ID. Can be specified multiple times for composite keys. Default: `_id` for BSON, `id` for JSON. | The output format is one extended JSON ID per line, which can be fed directly into `verify-ids --id-file` or `testsync --id-file`. ## verify-ids The `verify-ids` command fetches specific documents by ID from both the source and destination, optionally transforms the source documents, and compares them. It reports whether each document matches. Both connectors must support `GetByIds`. ```bash docker run -e 'DSYNCT_MODE=simple' \ markadiom/dsynct verify-ids \ --namespace \ --id-file ids.jsonl \ ``` To verify with a transformation applied to the source data before comparison: ```bash docker run -e 'DSYNCT_MODE=simple' \ -v "./transform.yaml:/transform.yaml" \ markadiom/dsynct verify-ids \ --namespace \ --id-file ids.jsonl \ --transform \ dsync-transform://transform.yaml ``` | Flag | Required | Description | | --------------------- | -------- | ----------------------------------------------------------------------------------------------- | | `--namespace` | Yes | The source namespace. | | `--dst-namespace` | No | The destination namespace. Defaults to the source namespace or the mapped namespace. | | `--id` | No | Document ID (string). Can be specified multiple times. For composite keys, use `--id-size`. | | `--jsonext-id` | No | Document ID in extended JSON format. | | `--id-file` | No | Path to a file containing extended JSON IDs, one per line. Compatible with `sample-ids` output. | | `--id-size` | No | Number of `--id` entries that form a single composite key. Default: `1`. | | `--transform` | No | Set if a transformer is provided after the two connectors. | | `--src-data-type` | No | Source data type. Inferred if not set. | | `--dst-data-type` | No | Destination data type. Inferred if not set. | | `--namespace-mapping` | No | Namespace mapping from source to destination. | | `--mapping-delimiter` | No | Delimiter for namespace mappings. Default: `:`. | At least one of `--id`, `--jsonext-id`, or `--id-file` must be provided. ## Typical Workflow Use `sample-ids` to collect IDs, then `verify-ids` to spot-check them: ```bash # 1. Sample IDs from source docker run -e 'DSYNCT_MODE=simple' \ markadiom/dsynct sample-ids \ --namespace mydb.mycollection \ --count 500 --output ids.jsonl \ # 2. Verify those IDs match between source and destination docker run -e 'DSYNCT_MODE=simple' \ markadiom/dsynct verify-ids \ --namespace mydb.mycollection \ --id-file ids.jsonl \ ``` For a full verification of all data, use the `verify` command instead: ```bash docker run -e 'DSYNCT_MODE=simple' \ markadiom/dsynct verify \ --namespace mydb.mycollection \ --parallelism 4 \ ``` ## Sync Tester (web-api) The web-api ships an interactive UI for spot-checking a sync without launching a full job. Start it the same way as the Transform Studio: ``` docker run -e DSYNCT_MODE=simple -p 8080:8080 markadiom/dsynct --host-port 0.0.0.0:8080 web-api --simple-only ``` Open the UI in a browser and navigate to `Tools > Sync Tester`. The tab exposes `Verify IDs` (read-only comparison, equivalent to the `verify-ids` command), `Test Sync`, and some previewing options. ### Connector requirements Not every action in this tab works for every connector. The constraints are: * **Sampling IDs** (the `Sample` menu) requires the connector being sampled to also be usable as a source. * **Verify IDs** requires both the source and destination connectors to be usable as sources and to support `GetByIds`. * **Test Sync** requires the source connector to support `GetByIds`. The destination connector only needs to be a valid sink. * Some preview/compare options also require the **Source Type** and **Destination Type** to be set explicitly (not left on `auto`); the UI will surface an error when this applies. ### Test Sync `Test Sync` runs a short, initial-sync style copy of a specific set of document IDs from source to destination through the regular worker pipeline. It is meant for end-to-end testing of a connector / transformer combination against a small, known set of documents before kicking off a real sync. To use it: 1. Pick a **Source** and **Destination** connector (and optionally a **Transformer**). 2. Fill in the **Namespace** (and **Destination/Transform Namespace** if it differs). 3. Optionally set **Source Type** and **Destination Type**; leave them on `auto` to infer. 4. Paste one extended JSON ID per line into the **IDs** textarea, or use the **Sample** menu to populate them from the source or destination. 5. Click **Test Sync**. The pipeline reads each ID from the source, applies the transformer (if enabled), and writes the resulting documents to the destination using the same code path as a real sync. Per-id write results are not surfaced by the pipeline. When a transformer is in use, namespace mapping is applied **before** the transform. This means the **Destination/Transform Namespace** field is the namespace handed to the transformer, and any namespace referenced inside the transformer's mapping config should be that post-mapping namespace, not the original source namespace. Use `Verify IDs` in the same tab afterwards to confirm the documents made it to the destination and match the expected payload. Note that the connector restrictions apply, so you may need to just check the destination database directly. # Web-API The `web-api` command starts an HTTP server that serves both a gRPC API and a single-page web UI for managing dsynct. Whereas the `dsynct run` command is scoped to a single migration workflow, the web-api is intended as a longer-lived control plane: it can list and start syncs, manage Temporal flows, register dynamic workers, and run interactive testing tools. To start the web-api: ```bash docker run --name dsyncwebapi \ -p 8080:8080 \ -e 'DSYNCT_MODE=simple' \ -e 'OTEL_EXPORTER_OTLP_ENDPOINT=http://:4317' \ markadiom/dsynct \ --host-port 0.0.0.0:8080 \ --otel \ web-api \ --temporal-host-port :7233 ``` The relevant flags for the `web-api` subcommand are: | Flag | Description | | ---------------------------- | ---------------------------------------------------------------------------------------------------------- | | `--temporal-host-port` | Address of the Temporal frontend, e.g. `temporal:7233`. Required unless `--simple-only` is set. | | `--temporal-namespace` | Temporal namespace to use. Default: `default`. | | `--simple-only` | Run without Temporal. Only Local Syncs and Tools are available; the Flows and Workers sections are hidden. | | `--cert-file` / `--key-file` | TLS certificate and key for serving HTTPS. | | `--no-gzip` | Disable gzip compression on the API. | Open the served address in a browser. The left-hand drawer contains four sections: **Progress**, **Syncs**, **Workers**, and **Tools**. ## Progress The Progress section shows live progress for any sync that the web-api is currently tracking. This includes both Local Syncs started from the Syncs tab and Temporal flows that have been added in the Flows tab. Use the dropdown at the top of the page to switch between tracked flows. For each selected flow the page shows: * A summary card with the workflow ID, current state, and (where applicable) a button to terminate the workflow or cancel the sync. * The latest activity errors, if any. * An **Initial Sync** card with per-namespace progress and controls to pause/unpause initial sync activities (overall or for a specific namespace). * A **Change Stream** card with controls to pause/unpause change stream and read-ahead activities. * An **Operation Log** that records the result of pause/unpause/terminate actions issued from the page. For local (non-Temporal) syncs, pause/unpause/terminate controls are not shown; only the cancel-sync button is available. ## Syncs The Syncs section is where you configure which syncs are tracked in the Progress view. It has two tabs: ### Local Syncs Local Syncs are non-Temporal syncs that run directly inside the web-api server process. They are intended for testing and for smaller projects where standing up a Temporal cluster would be overkill. Each local sync uses the same underlying pipeline as a worker, but progress is tracked in-memory by the web-api. To start a local sync: 1. Type a connector argument string into the **Run New Sync** input, for example `/dev/fakesource /dev/null`. Quoting follows shell-style rules. 2. Click **Run** to start the sync, or **Form** to open a guided form that exposes all `sync` flags and connector options. Started syncs are listed below with their current state (`Starting`, `Running`, `Done`, `Error`, or `Cancelled`) and the original arguments. Each row offers: * **View** -- jump to the Progress section for that sync. * **Cancel** -- cancel a running sync. * **Remove** -- remove a finished or errored sync from the list. ### Flows The Flows tab is the interface to Temporal. From here you can either start a brand-new flow or attach to one that already exists in Temporal: * **Run** -- type or build a `dsynct run`-style argument string (e.g. `--workflow-id myflow --queue-name dsync ...`) and submit. The web-api will start the workflow in Temporal and begin tracking it. The **Form** button opens a guided form for the same set of flags. * **Add Workflow** -- attach to an existing Temporal workflow. Click **List Workflow IDs** to fetch known workflow IDs from Temporal, then click **Add** next to any of them to start tracking it. The **Check Interval** and **Publish Interval** control how often the web-api polls Temporal for state and pushes updates to the UI. * **Active Flows** -- the list of currently tracked flows. Use **Remove** to stop tracking a flow (this does not affect the workflow on Temporal itself). Both Local Syncs and Flows feed into the Progress dropdown, so any sync started here can be monitored from the Progress section. ## Workers The Workers section is only shown when the web-api is running with Temporal (i.e. without `--simple-only`). It is used to control **dynamic workers**. Dynamic workers are dsynct worker processes started by long-running **node** processes that register themselves with the web-api over a coordinator API. Once registered, the web-api can ask a node to start or stop a worker using the same arguments you would pass to `dsynct worker`. This makes it possible to add or remove worker capacity from the UI without restarting any containers. ### Starting a node A node is started with the `node` command (run via the simple-mode entrypoint). Each node needs to know the Temporal address and the URL of the web-api that acts as its coordinator: ```bash docker run --name dsyncnode \ -e 'DSYNCT_MODE=simple' \ -e 'OTEL_EXPORTER_OTLP_ENDPOINT=http://:4317' \ markadiom/dsynct \ --host-port 0.0.0.0:8081 \ --otel \ node \ --temporal-host-port :7233 \ --coordinator-url http://:8080 \ --poll-interval 10s \ --heartbeat-timeout 60s ``` The relevant flags are: | Flag | Description | | ---------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `--temporal-host-port` | Address of the Temporal frontend. | | `--temporal-namespace` | Temporal namespace to use. Default: `default`. | | `--coordinator-url` | URL of the web-api that this node should poll for commands. If unset, the node still serves its node API directly but will not auto-register with a coordinator. | | `--poll-interval` | How often the node polls the coordinator for new commands. Default: `10s`. | | `--heartbeat-timeout` | The coordinator marks the node dead if it does not poll within this duration. Default: `60s`. | Run as many `node` processes as you need; each one becomes a separate row in the Workers section once it polls the coordinator. ### Managing workers from the UI Once at least one node is registered, the Workers section displays: * **Add Worker to All Nodes** -- type or build a `dsynct worker`-style argument string (e.g. `--queue-name myq /dev/fakesource /dev/null`) and click **Add to All** to start an identical worker on every registered node. The **Form** button opens a guided form with all `worker` flags. **Remove All Workers** stops every managed worker on every node. * **Nodes** -- one block per registered node, showing the node ID and last-seen timestamp. Inside each block: * A per-node argument input to start a worker only on that node. * A list of managed workers with their state (`Starting`, `Running`, `Stopping`, `Stopped`, `Error`), the original arguments, **Remove** to stop just that worker, and **Remove All Like This** to stop every worker (across all nodes) that was started with the same arguments. * Pending commands (add/remove) show their state (`pending`, `acked`, `error`) until they are processed by the node. Workers stopped by the coordinator continue to be reflected in the UI until they actually terminate, so there can be a short delay between clicking **Remove** and the row disappearing. {% hint style="warning" %} As with any other dsynct worker, every worker on the same `--queue-name` must be configured identically (same source/destination/transformer). Mixing configurations on the same queue will cause errors. {% endhint %} ## Tools The Tools section provides utilities for testing transformations and syncs **without** running a full job. There are two tabs. ### Transform Studio Transform Studio is an interactive way to iterate on a `dsync-transform` YAML config against sample documents. It is fully local to the web-api process -- nothing is written to any source or destination. Open `Tools > Transform Studio` and: 1. Paste your transform config into the **YML Config** field. 2. Set the **Namespace** to the source namespace the document belongs to. 3. Choose the **Src Type** and **Dst Type** (`DATA_TYPE_MONGO_BSON` or `DATA_TYPE_JSON_ID`). 4. Paste a sample document into **Payload** (extended JSON for BSON) and the document's ID into **Id** (also extended JSON, e.g. `{"_id": "123"}`). 5. Optionally tick **Show partial update** and supply a comma-separated list of fields to unset to test the `PARTIAL_UPDATE` mode. 6. Click **Run**. The Results panel shows the transformed output for each applicable mode (`Data`, `Insert`, `Update`, `Partial Update`, `Delete`). For details on the YAML format and CEL expressions, see [Data Transformations](/enterprise/running-dsynct/data-transformations). For converting between data types (e.g. string to BSON ObjectID), see [Transform Data Types](/enterprise/running-dsynct/data-types). Example extended JSON values: ``` {"$oid": "<24-character string>"} # Mongo Object ID {"$date": ""} # Date ``` ### Sync Tester The Sync Tester runs a small, initial-sync style copy of a specific set of document IDs from source to destination through the regular worker pipeline. It is meant for end-to-end testing of a connector / transformer combination against a small, known set of documents before kicking off a real sync. **All testing is local to the web-api -- it does not go through Temporal.** Open `Tools > Sync Tester`. The tab exposes `Verify IDs` (read-only comparison, equivalent to the `verify-ids` command), `Test Sync`, and some previewing options. #### Connector requirements Not every action in this tab works for every connector. The constraints are: * **Sampling IDs** (the `Sample` menu) requires the connector being sampled to also be usable as a source. * **Verify IDs** requires both the source and destination connectors to be usable as sources and to support `GetByIds`. * **Test Sync** requires the source connector to support `GetByIds`. The destination connector only needs to be a valid sink. * Some preview/compare options also require the **Source Type** and **Destination Type** to be set explicitly (not left on `auto`); the UI will surface an error when this applies. #### Test Sync To use Test Sync: 1. Pick a **Source** and **Destination** connector (and optionally a **Transformer**). 2. Fill in the **Namespace** (and **Destination/Transform Namespace** if it differs). 3. Optionally set **Source Type** and **Destination Type**; leave them on `auto` to infer. 4. Paste one extended JSON ID per line into the **IDs** textarea, or use the **Sample** menu to populate them from the source or destination. 5. Click **Test Sync**. The pipeline reads each ID from the source, applies the transformer (if enabled), and writes the resulting documents to the destination using the same code path as a real sync. Per-id write results are not surfaced by the pipeline. When a transformer is in use, namespace mapping is applied **before** the transform. This means the **Destination/Transform Namespace** field is the namespace handed to the transformer, and any namespace referenced inside the transformer's mapping config should be that post-mapping namespace, not the original source namespace. Use **Verify IDs** in the same tab afterwards to confirm the documents made it to the destination and match the expected payload. Note that the connector restrictions apply, so you may need to just check the destination database directly. For more details on the underlying commands, see [Verification](/enterprise/running-dsynct/verification). # Observability Dsynct exports logs and metrics via OpenTelemetry (OTel). This allows you to monitor migration progress, diagnose performance bottlenecks, and track change stream lag using tools like SigNoz, Grafana, or any OTel-compatible backend. ## Configuration Enable OpenTelemetry by passing the `--otel` flag and setting the `OTEL_EXPORTER_OTLP_ENDPOINT` environment variable to point at your OTel gRPC collector under the `app` part of the command: ```bash docker run \ -e 'OTEL_EXPORTER_OTLP_ENDPOINT=http://:4317' \ markadiom/dsynct app \ --otel \ \ ``` For simple mode (no Temporal) the options go before `sync`: ```bash docker run \ -e 'DSYNCT_MODE=simple' \ -e 'OTEL_EXPORTER_OTLP_ENDPOINT=http://:4317' \ markadiom/dsynct \ --otel \ sync \ \ ``` ### OTel Flags | Flag | Description | | ------------------------ | ------------------------------------------------------- | | `--otel` | Enable exporting logs and metrics to an OTel collector. | | `--otel-metric-interval` | Interval between metric pushes. Default: `10s`. | | `--otel-service-name` | Service name reported to OTel. Default: `dsynct`. | ### Logs When `--otel` is enabled, structured logs (JSON) are emitted both to stderr and to the OTel log collector. The log level can be controlled with `--log-level` (default: `INFO`). ## Metrics All metrics are emitted under the `dsync-flow` OTel meter. Metrics are labeled with attributes such as `namespace`, `success`, and `worker` to allow filtering and grouping. ### Common Attributes | Attribute | Description | | ----------- | ----------------------------------------------------------------------------------------------------------- | | `namespace` | The namespace (collection/table) being processed. | | `success` | `true` if the operation succeeded, `false` if it failed. | | `worker` | Identifies the worker type (e.g. `initial-sync`, `stream-changes`, `writer-0`, `transform-0`, `updates-0`). | | `index` | Stream partition index (for change stream gauges). | ### Initial Sync Metrics | Metric | Type | Unit | Description | | ---------------------- | --------- | --------- | ------------------------------------------------------------------------------------------------------ | | `dsynct.read` | Counter | documents | Total number of documents read from the source. | | `dsynct.written` | Counter | documents | Total number of documents written to the destination. | | `dsynct.list_data` | Histogram | ms | Latency of each `ListData` call to the source connector. | | `dsynct.write_data` | Histogram | ms | Latency of each `WriteData` call to the destination connector. | | `dsynct.get_transform` | Histogram | ms | Latency of each `GetTransform` call to the transformer. Only emitted when a transformer is configured. | ### Change Stream (CDC) Metrics | Metric | Type | Unit | Description | | ------------------------------ | --------- | ---------- | ------------------------------------------------------------------------------------------------------------ | | `dsynct.read` | Counter | events | Total number of change events read from the source. Shares the same counter as initial sync reads. | | `dsynct.written` | Counter | events | Total number of change events written to the destination. | | `dsynct.write_updates` | Histogram | ms | Latency of each `WriteUpdates` call to the destination connector. | | `dsynct.get_transform` | Histogram | ms | Latency of each `GetTransform` call during change stream processing. | | `dsynct.stream_read_gauge` | Gauge | events | Running total of change events read for a given stream partition. | | `dsynct.stream_written_gauge` | Gauge | events | Running total of change events written for a given stream partition. | | `dsynct.read_ahead_gauge` | Gauge | | The LSN (log sequence number) value reported by the source. Useful for tracking how far ahead the source is. | | `dsynct.last_event_time` | Gauge | ms (epoch) | Timestamp of the last change event processed, in milliseconds since epoch. | | `dsynct.since_last_event_time` | Gauge | ms | Time elapsed since the last change event was processed. Useful for detecting change stream lag. | ## Dashboards Pre-configured SigNoz dashboards are available in the [public repository](https://github.com/adiom-data/public/tree/main/kubernetes/system/signoz_dashboards). You can import them by following the [SigNoz import instructions](https://signoz.io/docs/dashboards/import-dashboard/). ### Key Things to Monitor * **Throughput**: Track `dsynct.read` and `dsynct.written` counters to monitor documents/events per second. * **Latency**: Use the `dsynct.list_data`, `dsynct.write_data`, and `dsynct.write_updates` histograms to identify slow operations. * **Change stream lag**: Monitor `dsynct.since_last_event_time` to detect if the destination is falling behind the source. A growing value indicates the CDC pipeline is not keeping up. * **Read-ahead**: The difference between `dsynct.stream_read_gauge` and `dsynct.stream_written_gauge` shows how many events have been read but not yet written, indicating backpressure. * **Errors**: Filter by `success=false` to isolate failed operations. # Azure Marketplace Deploy dsynct on AKS via the Azure Marketplace offering. {% hint style="info" %} Marketplace offering link: {% endhint %} The dsynct Azure Marketplace offering packages dsynct as an [Azure Kubernetes Service (AKS) marketplace application](https://learn.microsoft.com/en-us/azure/aks/aks-extensions-marketplace). One-click deploy provisions a [Web-API](/enterprise/running-dsynct/web-api) instance and, optionally, a number of dynamic worker **node** pods that register with it. Once the deployment is up, all further configuration -- starting syncs, attaching to Temporal flows, adding/removing workers, running test transformations -- happens through the Web-API browser UI.
## What gets deployed The offering installs the following workloads into your AKS cluster as a cluster extension: * **web-api** (1 replica) -- runs the dsynct web-api process and exposes the web UI plus the node coordinator API on port `8080`. Exposed either as a public `LoadBalancer` (external IP) or `ClusterIP` (internal-only, use `kubectl port-forward`). * **node** (0..N replicas; default `0`) -- runs dsynct as a long-running node process that polls the in-cluster web-api as its coordinator. Each replica becomes a row in the Web-API's [Workers](/enterprise/running-dsynct/web-api#workers) section, where workers can be added or removed without restarting any pods. * **temporal** (StatefulSet, 1 replica, optional) -- a single-pod `temporalio/temporal` dev server backed by SQLite on a persistent volume. Only deployed when you do **not** supply an existing external Temporal endpoint. If **Number of node workers** is `0`, no node pods are created. The web-api can still run **Local Syncs** and use the **Tools** section for testing -- this is the recommended starting configuration for evaluations and small projects. ## Configuration fields In addition to the usual AKS cluster parameters (VM size, count, Kubernetes version, OS SKU, auto-scaling, etc.) when creating a new cluster, the Azure Portal UI exposes the following dsynct-specific fields under **Application Details**: | Field | Description | | ----------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | **Number of node workers** | Number of dsynct `node` worker pods (0-10; default `0`). Each pod registers as a separate node and can manage one or more dynamic workers. Zero is valid -- the web-api can execute syncs locally. | | **External Temporal host:port (optional)** | `host:port` of an existing Temporal frontend (e.g. `my-temporal.example.com:7233`). Leave empty to deploy an in-cluster single-pod dev Temporal. | | **Web API service type** | `LoadBalancer` exposes the web UI on a public IP. `ClusterIP` keeps it internal (use `kubectl port-forward`). | | **Image tag** | Tag of the `markadiom/dsynct` image to deploy. | | **Temporal storage performance tier** | Storage class for the in-cluster Temporal SQLite database: Standard SSD, Premium SSD, or Premium SSD v2. Premium is recommended for production workloads (\~10x more IOPS). Only shown when no external Temporal is supplied. | | **Temporal storage size** | Disk capacity for the in-cluster Temporal SQLite database (5/10/50/100 GiB). Only shown when no external Temporal is supplied. | | **Container resources** | CPU and memory requests and limits applied to every pod, configured as four separate fields (CPU request, memory request, CPU limit, memory limit). Defaults: `1000m`/`512Mi` request, `2000m`/`1024Mi` limit. | | **OpenTelemetry collector endpoint (optional)** | gRPC OTLP endpoint for log/metric export (e.g. `http://otel-collector:4317`). Leave empty to disable OTel. | | **Enable shared config volume** | Provisions an Azure Files share (`ReadWriteMany`) mounted read-only at `/etc/dsynct` on the web-api pod and every node pod. Upload transformer YAML configs and other files post-deploy via the Azure Portal -- no rebuild or redeploy needed. | | **Shared volume size** | Capacity of the Azure Files share (1/5/10/50/100 GiB). Only shown when the shared volume is enabled. | ## After deploying ### Find the web-api address The Kubernetes resources are named after the **Cluster extension resource name** you chose during deployment. In the snippets below, replace `` with that value. If you chose `LoadBalancer`, watch for the external IP to be assigned: ```bash kubectl get service webapi- -w ``` Then open `http://:8080/` in your browser. If you chose `ClusterIP`, port-forward to access the UI from your laptop: ```bash kubectl port-forward svc/webapi- 8080:8080 ``` Then open `http://localhost:8080/`. ### Drive the deployment from the UI The browser UI contains everything you need to operate the deployment. See the [Web-API](/enterprise/running-dsynct/web-api) page for full details. In short: * **Progress** -- live progress for any sync that the web-api is tracking, with controls to pause/unpause/terminate where applicable. * **Syncs** -- start a Local Sync (runs inside the web-api pod) or kick off a Temporal flow, and attach to existing flows by workflow ID. * **Workers** -- visible only when Temporal is configured. Lists each node pod and lets you add or remove dynamic workers across all nodes or per-node from the UI. To add more node pods, update the deployment from the Azure Portal and increase **Number of node workers** (see [Scaling](#scaling)); the new pods will appear automatically once they poll the coordinator. * **Tools** -- Transform Studio (test transformer YAML against sample documents) and Sync Tester (do a small ID-bounded test sync, or compare specific IDs between source and destination). ### Editing the shared config volume When **Enable shared config volume** is on, the deployment creates a PersistentVolumeClaim named `shared-` backed by an Azure Files share (`ReadWriteMany`). The share is mounted read-only at `/etc/dsynct` on the web-api pod and every node pod, so you can reference files like transformer configs as `/etc/dsynct/transform.yaml` from any flag that takes a path. You can populate or edit files in the share at any time without redeploying. First find the storage account backing the PVC: ```bash kubectl get pv $(kubectl get pvc shared- -o jsonpath='{.spec.volumeName}') \ -o jsonpath='{.spec.csi.volumeAttributes.storageAccount}' ``` Then use the Azure Portal (Storage Accounts -> File shares), `az storage file upload`, `azcopy`, or an SMB mount from your laptop to drop files in. Every pod sees the change immediately. If a process reads a config file only at startup, restart the pods to pick up changes: ```bash kubectl rollout restart deploy/webapi- # Only if Number of node workers > 0: kubectl rollout restart deploy/node- ``` ## Bringing your own Temporal If you already operate a Temporal cluster, fill in the **External Temporal host:port** field with its endpoint (e.g. `my-temporal.example.com:7233`). The offering will skip deploying the in-cluster Temporal sidecar and point both the web-api and every node pod at your existing endpoint. Make sure the AKS cluster has network connectivity (and any required TLS/auth) to that Temporal frontend. When **External Temporal host:port** is left empty, an in-cluster dev Temporal is deployed for convenience. This is **not** suitable for production -- use it for evaluations and demos only, and bring your own Temporal for any real workload. ## Observability Set the **OpenTelemetry collector endpoint** field to the gRPC OTLP endpoint of any collector reachable from the cluster (e.g. `http://otel-collector:4317`) to enable metric and trace export. Leave it empty to disable OTel entirely. See [Observability](/enterprise/running-dsynct/observability) for details on the metrics and traces emitted by dsynct. ## Scaling The offering is deployed as an AKS cluster extension. To change configuration after the initial install (for example, to add or remove node worker pods), update the extension from the Azure Portal: 1. Open your AKS cluster in the Azure Portal. 2. Go to **Settings -> Extensions + applications**. 3. Select the dsynct extension and click **Update**. 4. Adjust **Number of node workers** (or any other field) and apply. You can also update the extension from the command line with [`az k8s-extension update`](https://learn.microsoft.com/en-us/cli/azure/k8s-extension#az-k8s-extension-update). New node pods register with the web-api on their next poll interval (default 10s); existing dynamic workers continue running uninterrupted. Use the [Workers](/enterprise/running-dsynct/web-api#workers) section in the UI to launch additional workers on the new nodes. ## Uninstalling Remove the deployment from the Azure Portal: 1. Open your AKS cluster in the Azure Portal. 2. Go to **Settings -> Extensions + applications**. 3. Select the dsynct extension and click **Uninstall**. You can also remove it from the command line with [`az k8s-extension delete`](https://learn.microsoft.com/en-us/cli/azure/k8s-extension#az-k8s-extension-delete). Persistent volumes (the in-cluster Temporal database and any Azure Files share) follow the StorageClass's reclaim policy and may outlive the extension. If you want the underlying data removed, delete the PVCs explicitly with `kubectl` or remove the storage account from the Azure Portal afterwards. # Deploy ## Deployment Models

Example dataflow between Cosmos DB and MongoDB

### Single binary Dsync will run wherever there's CPU and RAM. This can be a laptop, VM or a docker container. You can install dsync with any of the following methods: * Run as a docker container: `docker run markadiom/dsync` * Build from the [source](https://github.com/adiom-data/dsync) * Download the latest release in [GitHub](https://github.com/adiom-data/dsync/releases). Azure users can also leverage marketplace offerings for [Azure DocumentDB](https://azuremarketplace.microsoft.com/en-us/marketplace/apps/adiom.adiom_dsync_vm?tab=Overview) , [Cosmos DB NoSQL](https://azuremarketplace.microsoft.com/en-us/marketplace/apps/adiom.adiom_dsync_vm_dynamo?tab=Overview) and [other](https://marketplace.microsoft.com/en-us/product/virtual-machine/adiom.adiom_dsync_vm_general) migrations. {% hint style="info" %} You can use Homebrew to easily install Dsync on your Mac: ``` brew install adiom-data/homebrew-tap/dsync ``` {% endhint %} ### Multi-worker For large migrations where vertical scaling of dsync VM isn't feasible, and for data platforms requiring on-demand data mobility and continuous data replication, we support a horizontally scalable deployment of dsync with external Coordinator. Read more about it here: [Scalable Deployment](/enterprise/scalable-deployment) Azure users can leverage marketplace offering: . Read more about it in [Azure Marketplace](/enterprise/running-dsynct/azure-marketplace). ## Infrastructure Requirements Dsync doesn't store any in-transit data on-disk (only logs). As such CPU, RAM and network bandwidth are the core requirements. For small scale testing and development, 1 or 2 CPU with 2GB RAM is more than sufficient. For larger use cases, we recommend dedicating 4 CPU and 16 GB RAM, or 8 CPU and 32 GB RAM for a dsync instance, whether running as a single binary or in a multi-worker configuration. However Dsync is deployed, it must have direct network access to both the source and the destination. # Security ### General Dsync runs entirely within your secure perimeter. See [Deploy](/basics/deploy) and [Running Dsynct](/enterprise/running-dsynct) for deployment diagrams. Dsync doesn't "dial home" or contact any 3rd party services. Dsync uses network encryption to connect to databases when they support it and the connectors are configured appropriately. Dsync doesn't persist any transmitted data on-disk. ### Code We employ industry standard security code audits. Dsync Core is [Open Source](https://github.com/adiom-data/dsync/) and available for customer's own internal audits. For internal audits of Dsync Enterprise containers, please contact . # Features Here are some features of dsync that make our tools unique— safer, faster, and cheaper. ### Namespace Filtering and Remapping Allows us to selectively migrate data for specific databases or collections * Can be enabled with `--ns "db_name_1,db_name_2.col_name"` * You can also rename namespaces during a sync with `--ns "source_db.source_col:dst_db.dst_col".` * In some cases you can also rename without a fully qualified namespace (e.g. just the database), but it may not work properly with `--reverse` or `--mode CDC`. ### Data Integrity Check This check ensures that data remains unaltered and intact during the migration process. Dsync's data integrity check can be run separately from the data migration. Currently, dsync supports two data integrity check mechanisms: * Fast namespace counts with `--verify-quick-count` * Complete hash-based verification with `--verify` * See `dsync --help` for usage * Read [docs](https://github.com/adiom-data/dsync/blob/main/docs-dev/implementation-details/Integrity.md) for internal details The quick counts method can be used as a meaningful heuristic for most application. In most cases, it takes a trivial amount of time. The complete hash-based verification takes longer due to the need to read every single document. Dsync parallelizes the work where possible just like it does for the initial data copy. Additionally, if a mismatch is found, dsync will interrupt further checks and fail early (although that, of course, shouldn't happen during the normal course of operation). ### Resumability When restarting the flow during the initial data copy or Change Data Capture (CDC), dsync can safely resume from the last saved state. * Read [docs](https://github.com/adiom-data/dsync/blob/main/docs-dev/implementation-details/Resumability.md) for internal details * Read [docs](https://github.com/adiom-data/dsync/blob/main/docs-dev/implementation-details/TasksPartitioning.md) for granular resumability and partitioning ### Deletes Emulation Cosmos DB with MongoDB API currently doesn't support delete operations when using change streams, meaning that it doesn't emit deleted changes that occur. Dsync includes a workaround that captures and simulates delete operations from the source using a periodic index scan. * Can be turned on using `--cosmos-deletes-cdc` * See [docs](https://github.com/adiom-data/dsync/blob/main/docs-dev/implementation-details/CosmosDeletesEmu.md) for internal details **The destination namespace(s) must be empty or the pre-existing data might get deleted unless it exists on the source database.** ### Progress Report Dsync displays a detailed progress report of the sync process. This includes the time elapsed, number of namespaces synced, number of docs synced, number of tasks completed out of the total, percentage complete, and the throughput (docs/second). Dsync supports both CLI and Web-based progress report. * Web progress report is enabled by default and can be accessed at on the host where dsync is running. * CLI progress report can be enabled with `--progress --logfile dsync.log` dsync command-line options. Note that outputting logs into a logfile is a requirement. ### Load Level Dsync allows the user to specify the load level between "Low", "Medium", "High", or "Beast". The load level controls the number of threads employed for reading and writing data, with a higher load level corresponding to more threads. Generally, a lower level results in slower migration but less system impact, while a higher level uses more threads for faster migration at the cost of higher resource consumption. When load level is not specified, it will default to connector-specific settings. * Can be enabled with `--load-level LOAD_LEVEL` ### Rate Limiting Dsync supports write rate limiting to control the throughput of write operations to the destination. The rate limit effectively throttles reads on the source through back-pressure. You can configure the per-second rate limit using the `--write-rate-limit OPS_PER_SECOND` CLI option. The limit must exceed configured batch sizes. Rate limiting works on top of the load level and allows to leverage the optimal number of threads for your configuration without the fear of negatively impacting production traffic. For example, many connectors send large batches of operations during Initial Sync, and smaller batches during CDC. With the same number of writer threads, the load generated by Initial Sync is much higher. Rather than adjusting batch sizes and thread counts, rate limiter allows you to cap the overall ops/sec at a level that you're comfortable with. ### Reversing the flow Reverse the original flow by restarting dsync with the same parameters and added `--reverse`. This starts *a new flow* with inverted source and destination and in CDC mode, bypassing the initial data sync. When metadata store isn't specified, it will default to the *original* destination (the new source). `--reverse` is effectively a convenient shortcut for swapping source and destination, setting `-m`, and adding `--mode CDC`. **Reversal will respect the provided `--ns` option, but otherwise will replicate changes from ALL of the namespaces on the *****original***** destination (the new source), regardless of what was originally replicated.** Use the `--ns` option to limit which specific namespaces should be included in the reversal, if needed. ### Transformations Dsync supports data transformations via external transformer processes that dsync connects to over gRPC. Transformers can be implemented in any language (e.g. Go, Java, Python) as long as that language supports gRPC and the primitives necessary for transformations. For example, if the transformer process is running on localhost:8085 with no traffic encryption: ``` ./dsync source destination grpc://localhost:8085 --insecure ``` A sample identity transformer code can be found [here](https://github.com/adiom-data/dsync/blob/main/transform/identity.go), and the server that runs it is [here](https://github.com/adiom-data/dsync/blob/main/cmd/grpcexample/main.go). ### Other features You can run just `dsync` to see additional options - in particular to see the available connectors and connector specific parameters they accept. # How it works # Overview Learn about dsync internals on a high-level From the user's perspective, we aimed to make an otherwise complicated process as seamless and as easy as possible:

Live Migration process with dsync

But behind the scenes, dsync features specific machinery and orchestration to make data movement fast and reliable. Below are the key steps, see the [Glossary](/basics/how-it-works/glossary) for specific terms definitions:

High-level dsync architecture

Timeline

1. The Coordinator initiates the data flow by checking Source and Destination connectors' capabilities and *negotiating compatible options* (such as data format or database-specific functionality) 2. The Coordinator requests the Source connector to generate the best Read Plan for the initial data copy and create a checkpoint in the CDC stream (T0). The Read Plan consists of Tasks, each of which represents a particular subset of source data. Tasks are effectively units of read parallelization, and they are specific to source database type, since connectors take maximum advantage of native parallel scan features that are offered by source API. 3. The Source connector initiates the reading into the in-memory data channel using the Read Plan. Tasks are read in parallel according to the level or parallelism set for dsync. When the Read Plan is completed, the Source connector switches to the CDC mode starting to read the change stream from the source at the initial checkpoint (T0) onwards to ensure that no writes have been lost. 4. The Destination connector is reading the data from the in-memory data channel and writes it to the destination database as fast as possible according to the level or parallelism set for dsync. Each destination connector leverages database-specific optimizations to maximize writes efficiency. 5. During the process, the Source connector keeps the Coordinator updated on read progress, and uses an *in-band signaling mechanism* to communicate task boundaries, CDC progress and metadata updates to the Destination connector. Destination connector uses those signals to update the Coordinator on write progress. # Sync A look behind what goes on during the sync process: ### Planning Before a brand new sync process starts, dsync creates the read plan. A read plan is simply a set of tasks that need to be executed with some additional metadata. One or more tasks per namespace. Dsync tracks the completion of individual tasks to calculate progress and to provide a point of resume in case the process gets interrupted. Typically a read plan contains a set of tasks for initial data copy as well as the resume token for CDC. ### Initial data copy

Initial Sync

During the initial data copy stage dsync bulk loads the data from the source to the destination. Depending on the source data partitioning (how many tasks per namespace) and the configured level of parallelism (how many tasks are copied in parallel), the process can be very fast. ### Change stream

Change Stream

In order to capture data modifications made during and after the initial data copy process, a change stream mechanism is used to capture and track these modifications. Using resume tokens as a checkpoint for the last processed task allows for continuous synchronization and incremental data migration. ### Verification

Verification

After the data migration, a data integrity check ensures that the data has been accurately and completely transferred. # Glossary Commonly used glossary terms and definitions ### Flow Flow refers to the sequence of processes through which data moves from the source to the destination, started and managed by dsync. A flow is uniquely defined by the combination of the source, the destination and flow options (such as namespaces included). ### Read Plan A flow task plan essentially defines the flow status. The task plan requested from the source at the start of the flow includes the status of each task indicating whether they have been completed or not. The status is continiously updated throughout the flow execution. For resumable flows, the read plan is persisted in the metadata store. ### Connector Part of dsync. The source and destination connectors attach to each respective database and responds to commands from the coordinator. It manages connections to read data from the source database and write it to the destination database. ### Coordinator Part of dsync. The coordinator coordinates commands to manage the overall data flow between the connectors. ### Metadata store External database. The coordinator uses the metadata store to persist and retrieve the flow state and the task plan. This will default to the destination if not provided. # Limitations These are current limitations of dsync— take note that these areas may require manual intervention or special consideration during your migration process. ## General ### DDL and Index Replication Currently, dsync does not replicate Data Definition Language (DDL) statements or indexes from the source to the destination. This means that any database schema changes or index configurations must be handled separately to ensure they are correctly applied to the destination. ### Conflicts on the Destination Dsync does not manage conflicts on the destination system. If documents with the same \_id already exist on the destination, they will be overwritten by the incoming data from the source. As a result, users need to take care when migrating data to avoid unintended overwrites. ### Open Source Dsync version When used in conjunction with metadata store for flow resumability (the `"-m"` CLI option), Open Source Dsync version has a hard cap of 16MB on the read plan, that includes all generated read partitions on the source. This threshold is approximately equivalent to 2 billion records on the source. For large jobs, please use our [Scalable Deployment](/enterprise/scalable-deployment)version. Alternatively, you can try to reduce the number of generated partitions by choosing a larger partition size (`"--doc-partition"` CLI option) or by reducing the number of namespaces per dsync instance (`"-ns"` CLI option) ## MongoDB ### Partitioning Current data partitioning strategy doesn't support mixing different types of \_id values within any collection. ### Unique non-\_id indexes Existing non-\_id unique indexes on the destination can result in duplicate key errors when the values on the source are being actively changed. In MongoDB v6+ it's possible to temporarily disable the enforcement of uniqueness constraint during the data migration - see MongoDB [documentation](https://www.mongodb.com/docs/manual/reference/command/collMod/#change-index-properties). #### Updating Shard Key values Recent MongoDB versions allow updating shard key values. However, as of MongoDB 8.0, the implementation has several edge cases that may compromise logical data integrity on the destination. Generally, it's the safest to pause application workflows that update shard key values on a MongoDB source or a MongoDB destination for the duration of the migration. ## Cosmos DB (MongoDB API) ### Partitioning Current data partitioning for Cosmos DB only works for collections with \_id values of the ObjectId type. It works best when the values are evenly distributed through the lifetime of the collection, which is true for most applications. ### vCore Cosmos DB vCore Cosmos version only supported as a destination as it doesn't expose the ChangeStream API yet. ### Number of namespaces By default, we cap the number of namespaces to 8 when Cosmos DB is used as a source. This is due to a potential performance impact that we observed with more than 10-15 parallel ChangeStreams. You can use [namespace filtering](/basics/features#namespace-filtering) to specify which namespaces to sync, or you can use the `--cosmos-reader-max-namespaces` connector parameter to set the limit to a higher number. ### Replication lag reporting Since Cosmos DB change stream events don't expose any sequence number, we rely on an approximate client-generated sequencing mechanism. Due to its approximate nature, the reported replication lag might be greater than the actual value. ## Cassandra ### CDC Support CDC support is implemented using [Datastax Agent](https://github.com/datastax/cdc-apache-cassandra) and Apache Pulsar * Currently, the Agent (v2.3.9) works for Cassandra 3.11 and 4.0 but doesn't support versions 4.1 and 5.0 yet - there's an outstanding [PR](https://github.com/datastax/cdc-apache-cassandra/pull/206). * CDC for Cassandra 3.11 requires a force flush during the cutover window after the writes on the source have been stopped in order to ensure data integrity. # FAQs
What permissions do I need on the source or destination? Simple read on the source and write on the destination should suffice. For precise resume token creation on the source (highest consistency guarantee), write permission is necessary to inject a dummy record.
Can I write on the source while dsync is running? Yes.
Can I stop dsync and resume later? Yes, you can as long as you're using the -m option to store the metadata for the data flow, or using the Enterprise (Temporal-based) version. Then dsync will not repeat already completed tasks, and will automagically resume from the latest in-progress task. This is applicable both to the initial data copy and change-data-capture (CDC).
How can I restart the process from scratch? 1. Delete the data on the destination 2. Clean up dsync metadata by running it with the `--cleanup` option.
How fast is dsync? Dsync offers several different load levels that allow users to control the internal level of parallelism when reading data from the source and writing to the destination. Outside of that, the sync performance will depend on the following factors:\ 1\) Provisioned/available capacity of source and destination databases\ 2\) Network bandwidth\ 3\) Record/row/document size\ 4\) CPU allocated to dsync As reference data points, consider below: * 97GB migration with 55M documents and 200-500 writes/sec from Cosmos DB to MongoDB took 11 minutes * 334GB migration with 45M documents took \~1.5 hours
What security or auth options are supported? Dsync uses regular APIs for connecting to source and destination databases, such as [MongoDB Go Driver](https://www.mongodb.com/docs/drivers/go/current/) for MongoDB. All standard auth options are supported.
How do I start dsync in a highly-available configuration? You can run it as a docker container with a sensible restart policy.
Docker?
{% hint style="info" %} For any other questions not included in this FAQ section, contact us in [Discord](https://discord.gg/r4xzVfMQeU). {% endhint %} # Connectors Reference documentation for dsync connectors. # Cosmos DB NoSQL The Cosmos DB NoSQL connector is supported both as a source and a destination. It runs as a separate gRPC service that dsync/dsynct connects to, using the Java SDK optimized for bulk operations. ## Running the Connector Set the Cosmos DB account URL and key. These can be found under "Settings" -> "Keys" in the Azure Portal. ```bash export COSMOS_URI="..." export COSMOS_KEY="..." ``` ```bash docker run \ --network mynet --name cosmosnosqlconnector \ markadiom/cosmosnosqlconnector 8089 $COSMOS_URI $COSMOS_KEY ``` ## gRPC Configuration The connector exposes a gRPC endpoint. When specifying it as a source or destination, use the `grpc://` scheme with the `--insecure` flag (TLS is not enabled by default): ``` grpc://cosmosnosqlconnector:8089 --insecure ``` When running outside of Docker networking, replace the container name with the appropriate host (e.g. `grpc://localhost:8089`). ## Data Type The Cosmos DB NoSQL connector uses **JSON**. ## ID Format The document ID is composed of the shard key followed by the `id` field. The `id` field is always the last part of the ID. If `id` is itself part of the shard key, it is not duplicated. For example, with a shard key of `/region`: | `id` | `region` | Document ID | | ------- | ----------- | -------------------- | | `"123"` | `"us-east"` | `["us-east", "123"]` | With a shard key of `/id` (where `id` is the shard key): | `id` | Document ID | | ------- | ----------- | | `"123"` | `["123"]` | With a hierarchical shard key of `/region/tenant`: | `id` | `region` | `tenant` | Document ID | | ------- | ----------- | -------- | ---------------------------- | | `"123"` | `"us-east"` | `"acme"` | `["us-east", "acme", "123"]` | # MongoDB The MongoDB connector is supported both as a source and a destination. It is built into dsync/dsynct and connects directly using a MongoDB connection string. ## Configuration Provide a standard MongoDB connection string as the source or destination argument: ``` mongodb://username:password@host:port ``` This works with MongoDB Atlas (dedicated and serverless), self-managed installations, AWS DocumentDB, and Azure DocumentDB (Cosmos DB vCore). Note that change stream features may need to be explicitly enabled for MongoDB emulators unless you plan to run Initial Sync only. ## Data Type The MongoDB connector uses **BSON**. ## ID Format The ID is a single field called `_id`. By default, MongoDB uses ObjectID values for `_id`, but any BSON type is supported. # SQL Batch The SQL Batch connector is supported as a source only. It is built into dsync/dsynct and reads data from SQL databases using configurable queries. It supports SQL Server, PostgreSQL, and Oracle. ## Configuration Specify `sqlbatch` as the source with a `--config` flag pointing to a YAML configuration file: ``` sqlbatch --config=config.yml ``` ## Data Type The SQL Batch connector uses **JSON**. ## Config File The configuration file is a YAML file with the following top-level fields: ```yaml id: my-source driver: postgres connectionstring: "postgres://user:password@host:5432/dbname" mappings: - namespace: db.users query: "SELECT id, name, email FROM users" partitionquery: "SELECT id FROM users WHERE id % 4 = 0 ORDER BY id" cols: [id] limit: 1000 ``` ### Top-Level Fields | Field | Required | Description | | ------------------ | -------- | ----------------------------------------------------------------------- | | `id` | No | Identifier for the connector. Defaults to `"sql"` if not set. | | `driver` | Yes | SQL driver to use. Supported values: `sqlserver`, `postgres`, `oracle`. | | `connectionstring` | Yes | Connection string for the database. | | `mappings` | Yes | List of mapping definitions (see below). | ### Mapping Fields Each entry in `mappings` defines a namespace (analogous to a collection or table) that the connector will read. | Field | Required | Description | | ---------------- | -------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `namespace` | Yes | Name used to identify this data set (e.g. `db.users`). This is the namespace that dsync will use. | | `query` | Yes | SQL query that returns the data. Must include all key columns from `cols`. | | `partitionquery` | Yes | SQL query that returns partition boundary values. The columns must match `cols`. Results must be sorted in ascending order and contain no duplicates. Used to split the initial sync into parallel partitions. | | `cols` | Yes | List of key column names. Used for partitioning, ordering, and identifying rows. | | `limit` | Yes | Batch size for reading data. Must be at least 1. | | `countquery` | No | SQL query that returns a single count value. If omitted, defaults to `WITH QUERY AS () SELECT COUNT(*) from QUERY`. | | `nocount` | No | If `true`, skips the count query entirely. | | `decodejson` | No | List of column names whose values should be decoded from JSON strings into structured data. | | `fetchers` | No | Number of parallel fetchers for change streaming. Defaults to 1. | | `changes` | No | List of change tracking configurations for streaming updates (see below). | ### Change Tracking If you want the connector to stream ongoing changes (not just do an initial sync), configure the `changes` field. Each entry polls for changes using a query. The changes `query` receives the current cursor value as a single positional parameter. The placeholder syntax depends on the driver: | Driver | Placeholder | | ----------- | ----------- | | `sqlserver` | `@p1` | | `postgres` | `$1` | | `oracle` | `:1` | ```yaml mappings: - namespace: db.users query: "SELECT id, name, email FROM users" partitionquery: "SELECT id FROM users WHERE id % 4 = 0 ORDER BY id" cols: [id] limit: 1000 changes: - initialcursorquery: "SELECT MAX(updated_at) FROM users" query: "SELECT id, 'U', updated_at FROM users WHERE updated_at > $1 ORDER BY updated_at LIMIT 1000" interval: 5s ``` | Field | Required | Description | | -------------------- | -------- | ----------------------------------------------------------------------------------------------------------------- | | `initialcursorquery` | Yes | Query that returns a single value to use as the starting cursor. | | `query` | Yes | Query that returns changed rows (see format below). Use the driver-specific placeholder for the cursor parameter. | | `interval` | No | How long to wait before polling again when no changes are found. Defaults to `5s`. | #### Changes Query Format The changes `query` is critical to get right. It must return columns in this exact order: 1. **Key columns** -- the same columns listed in `cols`, used to identify which row changed. 2. **Update type** -- a string column: `"D"` for deletes, any other value (e.g. `"U"`, `"I"`) for upserts. 3. **Cursor value** -- a monotonically increasing value (e.g. a version number or timestamp) that becomes the input for the next poll cycle. For example, with `cols: [id]`, the query must return exactly 3 columns: `id`, update type, cursor value. The connector polls in a loop: it passes the current cursor as the placeholder parameter, reads all returned rows, advances the cursor to the last row's cursor value, and repeats until no more rows are returned. It then waits for `interval` before polling again. **Important considerations:** * The query should filter using `> cursor` (not `>=`) to avoid reprocessing the same row, unless the cursor value is set up to guarantee no overlap. * Results should be ordered by the cursor column in ascending order so that the connector advances through changes sequentially. * The number of rows returned per poll is bounded by `limit` from the mapping. If the query returns `limit` or more rows, the connector immediately polls again without waiting for `interval`. ## Full Example ```yaml id: my-sql-source driver: sqlserver connectionstring: "sqlserver://user:password@host:1433?database=mydb" mappings: - namespace: dbo.orders query: "SELECT order_id, customer_id, total, details FROM orders" partitionquery: "SELECT order_id FROM orders WHERE order_id % 8 = 0 ORDER BY order_id" cols: [order_id] limit: 5000 decodejson: [details] changes: - initialcursorquery: "SELECT CHANGE_TRACKING_CURRENT_VERSION()" query: "SELECT CT.order_id, CT.SYS_CHANGE_OPERATION, CT.SYS_CHANGE_VERSION FROM CHANGETABLE(CHANGES dbo.orders, @p1) AS CT ORDER BY CT.SYS_CHANGE_VERSION" interval: 10s - namespace: dbo.customers query: "SELECT customer_id, name, email FROM customers" partitionquery: "SELECT customer_id FROM customers WHERE customer_id % 4 = 0 ORDER BY customer_id" cols: [customer_id] limit: 2000 nocount: true ``` # Architecture Learn more about how it works The primary abstraction is a connector. Connectors are components that interface with the data stores that are responsible for fetching and writing data into them. Connectors conform to a defined gRPC interface and we can support any valid implementation. Connectors can be source only, destination only, or both. Dsync is software that coordinates data activities between connectors. For example, data movement/replication is supported from a connector that supports source capabilities into a connector that supports sink capabilities. Verification is supported for two connectors that support source capabilities. Dsync will ensure the specified connectors for any operation are compatible.
To support transforms, we also support a gRPC interface for transforms. Transforms are useful if data needs to be mapped from a source to destination. ``` TRANSFORM ^ | v SOURCE CONNECTOR <--> DSYNC <--> SINK CONNECTOR ``` ## Connector Interface For the specifics of the gRPC interface, refer to the [protobuf definitions](https://github.com/adiom-data/dsync/blob/main/proto/adiom/v1/adiom.proto). Connectors must define the capabilities they support via the `GetInfo` endpoint and basic information. For basic information, generally in type one would specify the underlying database, such as `MongoDB`. The `Id` is used to identify a connector is the same for resumability purposes. A connector must either provide a non nil definition for at least one of the `Source` or `Sink` capabilities to describe whether it supports being a source, destination, or both. Within each of these capabilities, a connector must specify which data types it supports. Currently, the only supported types are: * The `MONGO_BSON` type which is specifically a type that uses a BSON encoding and also by definition has an `_id` field that acts as the primary key * The `JSON_ID` type that has an `id` field as the primary key. This may not be true for all future types. ### Source Connector Source connectors by default support a single specified namespace, but can additionally be configured to declare support multiple namespaces or no specified namespaces. These are simply options to provide flexibility such as no specified namespace meaning "everything", but the interpretation is up to the connector. The definition of what string is a namespace is also up to the individual connector. For example, it could be a table name for DynamoDB, or a fully qualified `db.collection` identifier for MongoDB. Source connectors implement the `GetNamespaceMetadata`, `GeneratePlan`, `ListData`, and `StreamUpdates`, and `StreamLSN` endpoint. If it does not provide the lsn stream capability, then `StreamLSN` should just return success right away. `GetNamespaceMetadata` currently only asks for the size of the namespace, which could be an estimate as well. `GeneratePlan` is an important endpoint for specifying how the source can be partitioned. For the initial sync piece, a source could provide various ranges for query parallelism. For updates, a single partition may be returned to represent everything by not specifying a namespace. `ListData` and `StreamUpdates` are used to fetch the data for the initial sync and update plans respectively. `ListData` must preserve the output of the last cursor until the next cursor is used to account for failure scenarios. It is acceptable to expire the last cursor's validity after enough time has passed. On the updates side, the next cursor is used to update a checkpoint on the stream. Ideally every update should come with it so we can resume from that point onwards. ### Destination Connector Destination connectors implement the `WriteData` and `WriteUpdates` endpoints. These are both batch endpoints that should only return a success once the provided updates or the data is considered persisted. Furthermore reapplying the same update should not create duplicate data in the underlying data store. Under the hood they could both use the same implementation, the primary difference is that `WriteData` may provide an opportunity for optimizing a pure insert/upsert command. ## Transform Interface For the specifics of the gRPC interface, refer to the [protobuf definitions](https://github.com/adiom-data/dsync/blob/main/proto/adiom/v1/adiom.proto). `GetTransformInfo` is implemented to determine what data type to what data type mappings are supported. At the very least this could be just the same data type to itself. `GetTransform` is the actual endpoint that performs the transform. It basically should apply the requested transform to both the `data` and `updates` whichever are present and may also update the `namespace`. ## Signaling Protocol In order to effectively decouple the reading and writing responsibilities, connectors implement an in-band signaling protocol to communicate task boundaries, CDC progress and metadata updates. For example: * When a task is fully read, the source connector sends a special message to let the destination connector know that there will be no more data coming for that task. When the destination connector sees that message, it can assert that once all pending data for the task is written to the destination, it's safe to notify the Coordinator that the task is fully complete. * During CDC, the source connector sends periodic "barrier" messages that contain the current stream offset or token ("checkpoint"). Upon encountering a barrier message, the destination connector can assert that once all pending change events are processed, it's safe to communicate the checkpoint to the Coordinator. For all connectors using the gRPC interface, the signaling protocol is implemented automatically. # Verification Dsync provides multiple verification options. Using the `dsync verify` command triggers the verification mode. By default this uses a Merkle Search Tree to load up all items on both connectors and compare their xxhash and print out any differences it sees. Since this can be memory intensive there is an option to specify the number of partitions to filter out to only comparing 1/n. It takes time to load the initial items, and then it will also run indefinitely tailing the CDC updates. Using the `--simple` option will use a simple verifier that only tails the CDC updates of the 2 connectors. It periodically checks any "stale" updates and compares those. This relies on both data sources starting at an already stable state and basically ensures that updates on one side show up in the updates of the other. The first few differences may be discarded as it is possible when starting this verifier we just missed an update from one connector but got the update in the other. Any data that is updating fast enough won't be compared as the condition to compare is a stale update. ## Legacy Verification Specifying the `--verify` option is a legacy feature that only works for some pairs of connectors. There are 2 modes supported here. By default it runs a hash-based data validation using the source read plan. It assumes that the sink is also a source and can understand the source plan, which is not guaranteed thus not all pairs of connectors support this feature. It will do an xxhash and xor it all across all items in each partition and compare. `--verify-quick-count` will simplify the above so that it does a query for the approximate number of documents and see if they are the same. As long as both connectors support `GetNamespaceMetadata` this would work. # Resumability The dsync tool supports resumability by default assuming connectors are implemented properly. Dsync uses multiple writers but can track write completion by relying on a special message that all writers must see and so when the last writer sees that message we will know that all writes prior to it have completed. Resumability is currently tracked by using a metadata database (currently supports MongoDB) which can be specified with the `-m` option. ``` METADATA DB ^ | v SOURCE CONNECTOR <--> DSYNC <--> SINK CONNECTOR ``` Initial sync resumability is at a partition level. To make this more clear, let's look at an example. `GeneratePlan` returned 20 partitions. We have completed 12 of these, and are almost completing 4 more. Then we interrupt the process and resume. We will have to restart the 4 that were almost complete from the beginning so we will resume with 8 more tasks to complete. Partition completion is tracked only after the last write (not read) of that partition's contents. The updates/CDC resumability is based on the last saved next cursor and a periodic event. Every so often dsync will persist the last known next cursor by emitting an event and waiting on all writers to see that event before persisting so we can resume from that point if there is an interruption.