Databases

A Database source flow extracts data from a relational database and loads it into a cloud data warehouse or storage target. Use this flow type when your source is a database system, such as MySQL, PostgreSQL, or Oracle rather than a SaaS application or file storage.

Supported database connectors include MySQL, PostgreSQL, Oracle, SQL Server, BigQuery (as source), Snowflake (as source), Amazon Redshift (as source), MongoDB, MariaDB, Teradata, and Vertica.

For the full list of supported connectors, refer to Databases in the Sources section.

Example: A company wants to replicate their PostgreSQL production database into Snowflake for analytics and reporting without impacting the source system.

Flow: PostgreSQL > Source to Target > Snowflake

Before you begin

Before creating a Database source flow, ensure you have the following:

• An active Data Integration account.
• Valid database credentials (hostname, port, username, password) and network access from Data Integration to your database. Refer to Database connectivity options for firewall and IP allowlisting requirements.
• A configured database connection in Data Integration. Refer to Creating a connection for details.
• Access to a supported cloud data warehouse or storage target.

Step 1: Create a Data Flow

1. Navigate to the Data Integration Console.
2. Click Data Flows in the left-hand menu.
3. Click Create Data Flow and select Source to Target Data Flow, or open an existing data flow.

Step 2: Set up the source

1. In the Source tab, select your database connector. For example, select MySQL, PostgreSQL, or Oracle.
2. Select or create a connection. Existing connections appear in the Connections drop-down menu. To add a new one, click New Connection.

Step 3: Select the extraction mode

Database source flows support three extraction modes. Select the one that fits your data volume, update frequency, and transformation requirements.

Extraction mode	Best for
Standard extraction	Regular batch loads from one or more tables. Supports incremental extraction by timestamp or ID.
Change Data Capture (CDC)	High-volume databases with frequent updates. Captures only changed records in real time with minimal load on the source.
Custom Query	When you need to join tables, filter rows, or transform data during extraction using a SQL SELECT statement.

Standard extraction

Standard extraction loads data from one or more source tables on a scheduled basis. You can select multiple tables in a single flow. Each table maps to a corresponding target table. Data Integration automatically generates target tables and columns based on the source schema.

After selecting a table, configure the extraction method:

• All: Extracts the complete table on every run. Use for small or infrequently changing tables. |
• Incremental: Extracts only records added or modified since the last run, based on a timestamp or ID field. Use to minimize data volume on large tables. |

For incremental extraction, select the incremental field, such as created_at, updated_at, or auto-increment ID column, and configure the start and end date range. Refer to Incremental extraction settings below.

Change Data Capture (CDC)

CDC tracks changes in the source database by monitoring transaction logs. Instead of querying the full table on every run, Data Integration reads only INSERT, UPDATE, and DELETE operations recorded since the last run, then applies them to the target. This makes CDC highly efficient for large, frequently updated databases.

CDC requires the source database to have binary logging or change tracking enabled. For setup requirements, refer to the documentation for your specific connector.

note

CDC Data Flows use an Activate step instead of Run. After configuring the flow, click Activate to start continuous change capture. If you make configuration changes to an active CDC flow, you must reactivate it for the changes to take effect.

Custom Query

Custom Query lets you write a SQL SELECT statement to define exactly what data Data Integration extracts. Use this mode when the extraction requires joining tables, filtering rows, creating calculated fields, or flattening nested data, transformations that cannot be done by selecting a table directly.

Custom Query supports a single target table per flow.

note

• Only SELECT statements are supported. Write operations (INSERT, UPDATE, DELETE, MERGE) are not permitted.
• Write the query in the SQL dialect of your source database (for example, MySQL SQL, PostgreSQL, Snowflake SQL).
• Custom Query requires only read permissions on the queried tables.

To configure a Custom Query:

1. Select Custom Query as the extraction mode.
2. Enter your SQL query in the editor. Expand the editor to open the full SQL view.
3. Click Run Query to validate the SQL and preview the results.
4. Click Save & Close.
5. Define the target table in the Loading To field.
6. Click Run Mapping to validate the query and fetch column definitions.
7. Click Next to continue.

After running mapping, select the extraction method:

• All: Extracts the full query result on every run.
• Incremental: Extracts only new or updated records. Requires an incremental field in the query.

Incremental extraction settings

When using incremental extraction (Standard extraction or Custom Query), configure the following settings:

Setting	Description
Start date	The beginning of the time range from which to extract data.
End date	The end of the time range. Leave empty to extract data up to the current run time.
Days back	Extends the start date backwards by the specified number of days. Use to capture late-arriving or backdated records.
Time zone offset	Aligns the extraction window with the local time zone of the source system when the end date is left empty.

note

• After each successful run, Data Integration automatically advances the Start Date to the previous run's End Date, ensuring the next run picks up from where the last one stopped.
• Data Integration does not advance the Start Date if a run fails. To override this behavior and advance the date even on failure, go to More Options and enable the corresponding checkbox. This is not recommended, as it may cause data gaps.

Step 4: Select a target

In the Target tab, select the cloud destination where Data Integration loads the extracted data.

Data Integration supports all major cloud data warehouses and cloud file storage systems as targets, including Snowflake, BigQuery, Amazon Redshift, Azure Synapse Analytics, Databricks SQL, and others.

Select or create a connection for your target. Then define the Database, Schema, and Target Table where the data will be stored. Data Integration automatically detects available databases and schemas.

Loading modes

Choose how Data Integration writes data into the target table:

• Overwrite: Replaces all existing data in the target table with data from the current run. Creates the table automatically if it does not exist. Full refreshes, reference tables, or initial loads.
• Append only: Adds new records without modifying existing rows. Audit logs, event tables, or append-only datasets.
• Upsert-Merge: Inserts new records and updates existing ones based on a defined merge key. Keeping the target in sync with the source for tables that receive updates.

If you select Upsert-Merge, define your merge key in the Schema tab (Step 5).

Advanced target settings (Snowflake)

When loading into Snowflake, the following advanced options are available under Advanced Settings:

• Truncate columns: Truncates VARCHAR values that exceed the defined column length, preventing overflow errors.
• Replace invalid UTF-8 characters: Replaces invalid UTF-8 characters with the Unicode replacement character to prevent processing errors.
• Add Data Integration metadata: Adds metadata columns to the target table: Data Flow_last_update, Data Flow_Data Flow_id, and Data Flow_run_id. Use these for auditing and lineage tracking.
• Custom File Zone: Specifies a custom staging file zone before loading into Snowflake. Useful for managing large files or custom storage configurations.

note

For file storage targets, specify the bucket and file path where Data Integration stores the data. The connection's default bucket is used if you do not specify one.

Step 5: Configure the schema

The Schema tab has two levels of configuration Flow settings that apply to all tables, and Individual table settings that you configure individually for each table.

note

For Custom Query flows, the Schema tab opens directly at the column mapping. There is no tables view since Custom Query uses a single target table.

Flow settings

These controls are available in the Schema tab and apply across all tables in the flow.

Extraction Mode

Click Extraction Mode to switch between CDC, Standard extraction, and Custom Query without leaving the Schema tab.

Tables Definitions

Click Tables Definitions to set flow-level source and target defaults. Changes here apply to all tables unless overridden at the table level.

Advanced Source Definitions

• Replace invalid characters for target tables/columns names: Automatically replaces characters in source table and column names that are invalid in the target system. Enable this when your source uses special characters or spaces that the target warehouse does not support.

Advanced Target Definitions

• Table prefix (optional): Add a character or phrase to the beginning of every target table name. For example, entering ODS renames orders to ODS_orders.
• Default loading mode: Set the loading mode applied to all tables. Overwrite, Append only, or Upsert-Merge. Per-table settings override this default.
• Merge method: When using Upsert-Merge, select the merge strategy: Switch - Merge or other supported options.
• Filter logical key duplication between files: Filters out duplicate records in the current source pull before loading. Enable only when duplicates are expected in the source but not in the target table.

Click Apply Changes to save.

Bulk Actions

Click Bulk Actions to configure properties across multiple tables at once. The button becomes active after you select two or more tables.

1. Choose a selection scope: All Tables in Selected Schemas or All Tables.
2. Configure the properties below, then confirm. This action is irreversible.

Available properties:

• Extract method (Standard extraction only): Set all selected tables to All or Incremental with a shared incremental field. The field type (Timestamp, Date, Running Number, Epoch) is detected automatically.
• Set time period: Define a custom date range for all tables using date or timestamp-based incremental fields.
• CDC initial migration options: For CDC tables, choose whether to Overwrite or Merge existing target data during initial migration.
• Loading mode (not applicable for CDC): Set all tables to Overwrite, Append, or Upsert-Merge.
• Create a calculated column: Add a calculated column to all selected tables. If a column with the same name already exists, it is overwritten.

Reload Metadata

Click Reload Metadata to refresh schema metadata after changes in the source database, such as added columns or renamed tables.

• Reload metadata for selected schema: Refreshes metadata for a specific schema.
• Reload metadata for all schemas: Updates metadata across all schemas in the flow.

---

Individual table settings

Click any table name in the tables view to open its individual configuration. A side panel opens with three tabs where you configure that table's columns, extraction behavior, and load behavior.

Mapping

For Standard extraction and CDC flows, click Auto-Mapping to detect column names, data types, and modes from the source table automatically. For Custom Query flows, columns are already populated from the Run Mapping result in Step 3.

You can edit the following for each column:

• Target field name: Rename the column as it appears in the target table.
• Data type: Change or cast the data type during loading.
• Mode: Controls how the column handles null values:
  • Nullable: The column can contain null values. Use for optional fields.
  • Required: The column must always contain a value. Use for IDs, timestamps, and mandatory fields.
  • Repeated: The column stores multiple values per row as an array. Supported in BigQuery.
• Primary key (match key): Mark one or more fields as the merge key for Upsert-Merge flows. To create a composite key, select multiple fields, the combination of those values uniquely identifies each row.
• Cluster key: Select columns to organize data for query performance. Data Integration arranges cluster keys in descending order. Supported in Snowflake, BigQuery, and Redshift.
• Expression: Write SQL to create a calculated column or transform an existing field using target warehouse syntax. For more information, refer to Targets.

note

• Data Integration does not include column descriptions in the target metadata. Document field descriptions separately if your data governance process requires them.
• Column names cannot contain dots (.). Dots cause errors during data processing. Rename any affected fields before running the flow.

Table Source Settings

Configure extraction behavior for this specific table. These settings override the flow-level defaults set in Tables Definitions.

• Extraction method: Switch between All and Incremental at the table level.
• Update incremental date range on failures: Controls whether the incremental date range advances when an extraction fails. Enabling this may cause data gaps.
• Interval chunk size: Splits the extraction window into smaller time chunks per database call. Use when large queries time out or exceed rate limits.
• Filter expression: Enter a SQL filter expression to restrict which rows are extracted from this table.

Table Target Settings

Configure how data is loaded for this specific table into the target.

note

Table Target Settings vary by target platform. The options below apply to Snowflake. Refer to the documentation for your specific target for a full list of available settings.

• Filter logical key duplication between files: Filters duplicate logical keys across files before loading, preventing duplicate rows from reaching the target.
• Enforce masking policy: Applies a data masking policy to sensitive columns. Refer to Enforce masking policy.
• Support escape character: Enables escape character handling for special characters in column values. Refer to Support escape character.

Step 6: Schedule and run the Flow

How you run the flow depends on the extraction mode you selected:

• Standard extraction and Custom Query: Click Run to execute immediately. To schedule recurring runs, view below.
• CDC: Click Activate instead of Run. The flow runs continuously, capturing changes as they occur. Scheduling is not applicable for CDC flows.

Schedule the Flow

Click Schedule Me in the Settings tab to set up automatic runs.

note

Scheduling intervals by plan:

• Starter: minimum 60-minute interval
• Professional: minimum 15-minute interval; includes custom CRON expressions
• Enterprise: minimum 5-minute interval; includes custom CRON expressions

Using a CRON expression

Professional and Enterprise plans support custom CRON scheduling using Quartz format. The pattern has seven space-separated fields:

<second> <minute> <hour> <day of month> <month> <day of week> <year>

Enter your expression under the Custom tab. Use last instead of L for last-day-of-month syntax. For example, to run on the last day of every month at 22:30:

0 30 22 last * ? *

Extended execution time

For multi-table flows, Data Integration automatically extends execution time up to 48 hours for large RDBMS tables and predefined large API reports. You can set a custom timeout limit in the Settings tab to override this default.

Set up notifications

In the Settings tab, enter your email address to receive alerts for run failures, warnings, or runtime threshold breaches.

note

• Each table in a multi-table flow runs as a background process. You receive a separate alert for each failed table.
• Enable On Warning to also receive alerts when a run returns no data.
• To receive alerts in Slack, create a dedicated Slack channel and connect it to Data Integration's email alert system.

note

If you make configuration changes to an active CDC flow, such as adding tables or modifying column mappings. You must click Activate again to apply the changes. The flow pauses during reactivation.

Monitor the Flow

After the flow runs, open the Activities tab in the right-hand menu to review run status. For each run, Activities shows the status, duration, data size, and BDU cost per table, along with any warnings or errors. Select a run to view full run details, including source, target, size, and the date range processed. Click Download Log for the complete run log.

For CDC flows, use the Activities tab and the Data Integration dashboard (Activity Stats and Runs Segmentation) to confirm the flow is actively capturing changes and that no tables are lagging.

note

A Data Flow supports up to 500 tables.

Deployments

Data Integration supports deploying Database source flow configurations from one environment to another.

During deployment, note the following behavior:

• Data Flow type and status: You can view the flow type and the current status of the source environment and the intended status of the target environment in the deployment panel.
• Original status retained: Database Data Flows keep their original status when deployed. CDC Data Flows are deployed with a Disabled status.
• Existing flows in the target environment: If the Data Flow already exists in the target environment, Data Integration updates its configuration without changing its current status. Active flows remain active and are not re-validated.

note

When deploying to an environment where re-validation is not required, manually reactivate CDC Data Flows in the target environment to confirm correct behavior after deployment.