Create continuous replication pipeline - Precisely Data Integrity Suite

Create a continuous replication pipeline

To create a continuous replication pipeline:

1. On the main navigation menu, select Integration > Continuous Replication.
2. Select the project that you want to use to create your continuous replication pipeline.
3. Select +Create Pipeline to open the Create Pipeline wizard.
4. Add the information for the name and description for a continuous replication pipeline.
5. Select Next.
6. Select a connection for Source and Target connections.
   Note: If Oracle is selected for Source connection, you need to create a metabase.
7. Select Next.
8. Select a schema and associated tables, then select Next.
9. Map the tables and select Next.
10. View the information about the continuous pipeline, see the next step.
11. Select Finish to return back to the Continuous replication pipelines page and view the list.

To view your list of continuous replication pipelines from the Replication Pipelines page, go to Integration > Continuous Replication > Select the project to view its pipeline

Step 1: General setup

• Replication pipeline: Provide name of the replication pipeline. Every project contains one or more replication pipeline. You configure the pipeline properties that, when the pipeline is run or started, move source data to a target database or file system in bulk or run data capture and replication processes.
• Description: Additional information about the replication pipeline.
• Type: Choose the replication type:
  • Synchronize: Copies a snapshot of a set of source tables from the database source and distributes it to the target. Data changed during the copy process is captured. When the copy process finishes, a change-based replication begins, applying all changes made during the copy up through the time the copy finished. Thereafter, changes made to the source data are detected, captured, and replicated to the Kafka or Snowflake target, synchronizing the table column data.
  • Replicate: Detects changes made to data, captures the changes, and applies them to the Kafka or Snowflake target.
  • Copy: Copies data extracted from the database source and distributes it to the target. Not supported for the pipeline using Db2 for IBM i datasource connections.
  • Audit: Provides an audit log of all changes made to the source without actually altering the target dataset. Changes to the source data are detected and captured, but rather being applied to the target, information about each change is simply recorded in the target dataset. This includes details such as whether the change on the source was an insert, update, or delete, as well as timestamp of when the change occurred.
    Note:

    • This type of replication is available only for journal mapping.
    • Target tables in audit pipelines should not have primary keys. If primary keys are required, they should be based on values that are NOT part of the source table, such as a unique row metadata value. This is because, in audit pipelines, rows are always inserted into the target table and never updated.
• Replication Options:
  • Transaction error mode: Select what happens when an error occurs while the replication pipeline transfers data to the target and the error does not affect the connection to the target server.
    • Delete from queue and continue: Processing continues despite the error. The statement and record that caused the error are copied to the log then deleted from the replication backlog tables.
    • Shut down apply component: For the errored replication pipeline transaction, transferring data to the target is stopped and rolled back and all connections are shut down. The source server's replication backlog tables remain in the state that the last successful transfer left them.
  • Conflict resolution between source and target (DBMS only): Depending on how you want to manage and resolve conflicts between data in the source and target found during data replication activities:
    • Source data is correct. Update the target: Selected by default.
    • Target data is correct. Do not update the target
• Copy Options:
  • Copy method
    • Use load mode when possible: This option is available exclusively for SQL Server, Oracle, and Teradata, and utilizes the database's fast load utility to enhance data loading performance. Additionally, it allows data to be copied in batches of 100,000 rows.
      Note: The loader files are automatically cleaned up when the load process is successful. However, they will remain if an error occurs during the load or if debug mode is enabled. In the event of an error, the loader files will be cleaned up once the copy operation runs successfully during the next execution.
  • Error mode: Select the type of error mode to use when a non-recoverable error occurs for a request to the target connection.
    • Stop run (default): Stops the request.
    • Skip record: Continues to the next target row.
    • Next table: Continues to the next table in the request.
  • Source isolation level: Select the transaction isolation level to set at the source server.
    • Read committed(default): Extracts only committed changes. Extracted rows can be updated by other concurrent transactions, however the extracted data will contain only committed changes.
    • Read uncommitted: The extracted data may include uncommitted changes from running transactions. This option may improve processing performance.
    • Repeatable read: Extracts only committed changes so that other concurrent transactions cannot update the extracted rows during the extract transaction. Other concurrent transactions may add new rows that satisfy a search condition of the extraction, however the extract transaction will not include those rows.
    • Serializable: Extracts only committed changes so that other concurrent transactions cannot update extracted rows or add new rows that satisfy a search condition during the extract transaction.
  • Source locking: Select the table locking approach to use at the source during the copy process.
    • By individual table (default): The process accesses and locks, one at a time, the source tables selected for copying, then commits and unlocks each table after processing.
    • All tables: The process accesses and locks, one at a time, the source tables selected for copying, then commits and unlocks the tables after processing all the tables.
  • Source table retrieval order: Select the order to copy data tables.
    • Retrieve in any order (default): Copies tables in the order they are received.
    • Retrieve in order of mappings: Copies parent tables before their child tables. Select this option when the tables being copied have referential integrity constraints.
  • Conflict resolution between source and target (DBMS only) : Select one of the following options, depending on how you want to manage and resolve conflicts between data in the source and target found during data copy activities.
    • Clear the target table before copying. Records not on the source will be lost. Selected by default.
    • Only copy records that are not on the target.
    • Source record is correct. Update the target.
    • Target record is correct. Only update the mapped columns.

Step 2: Connections

• Add at least one data connection for each pipeline. Connections must be assigned to at least one pipeline for source connections or two pipelines for target connections.
• Add pipelines and associate each pipeline with at least one source connection and one target connection.
• Run or start your Data Integration project from this step.

Use data connections to access both source and target systems. Connections configured to access Kafka are only valid for Kafka topics, Data Streams, and Change Data Capture operations.

Add a source connection

Be aware of the following when setting up data connections:

• When adding data connections for relational databases, ensure that the relational source files are properly configured before you run a project.

1. On the main navigation menu, select Integration > Continuous Replication.
2. Select the project that you want to use to create your continuous replication pipeline.
3. Select +Create Pipeline to open the Create Pipeline wizard.
4. As a second step in connections, select Add Source. The New Replication Connection dialog opens.
5. For replication connection, enter a unique name for the data connection.
   Note: If you enter a name that already exists in the repository, you cannot save the connection. The name cannot exceed 200 characters. For replication data connections, spaces in the name are not supported and hyphens ( - ) are the only special characters supported.
6. For Description, enter a brief description of the data connection.
7. For connection type, select the dropdown list and select the type of connection you are adding. If you select a DBMS, specify the associated data Access method.
8. Specify the properties for your selected connection.
9. Select Test to verify that your credentials are valid for the properties you entered.
   Note: If the Test button is unavailable, make sure you have entered values in all required fields.
10. Select Save to add the new connection.

Add a target connection

Specify the target database, file system, data warehouse, or Kafka stream to receive replicated data.

• Pipelines for replication projects are configured to copy or replicate data changes to a single Kafka or Snowflake target.
• You define Apache Kafka properties such as row properties, message format type (CSV or JSON), specify large object types (LOB), and whether to write to a new topic you create on the Kafka server or use an existing topic.
• A Kafka target is a producer, publishing messages to a Kafka topic. Topics are categories where the Kafka cluster stores streams of records.
• For Snowflake targets, you select a warehouse, database, and schema with the option to create a target table if it does not exist. You also specify the batch criteria for applying records to the target.

To add a target connection, follow these steps:

1. On the main navigation menu, select Integration > Continuous Replication.
2. Select the project that you want to use to create your continuous replication pipeline.
3. Select +Create Pipeline to open the Create Pipeline wizard.
4. As a second step in connections, select Add Target. The New Replication Connection dialog opens.
5. For replication connection, enter a unique name for the data connection.
   Note: If you enter a name that already exists in the repository, you cannot save the connection. The name cannot exceed 200 characters. For replication data connections, spaces in the name are not supported and hyphens ( - ) are the only special characters supported.
6. For Description, enter a brief description of the data connection.
7. For connection type, select the dropdown list and select the type of connection you are adding. If you select a DBMS, specify the associated data Access method.
8. Specify the properties for your selected connection. For Azure SQL Database target connections, the database user defined in the connection must have SELECT, INSERT, UPDATE, DELETE, and CREATE permissions at a minimum.
9. Select Test to verify that your credentials are valid for the properties you entered.
   Note: If the Test button is unavailable, make sure you have entered values in all required fields.
10. Select Save to add the new connection.
11. Select Next.

Configure Snowflake as a target

When selecting Snowflake as a target connection for continuous replication pipeline, the following fields need to be configured:

• Storage to use for staging: Storage locations such as cloud storage location is utilized for loading data from files into Snowflake tables.

  Storage type	Fields
  Amazon S3	Method: Specify the authentication method and possible values are key or session token. When selecting Key as the authentication method, the following fields must be filled: AWS access Key: Specifies the access key required for authentication. AWS secret key: Specifies the secret key required for authentication. When selecting Session Token as the authentication method, the following field must be filled: Session token: Specifies the session token required for authentication. AWS access Key: Specifies the access key required for authentication. AWS secret key: Specifies the secret key required for authentication. When selecting Key Vault as the authentication method, the following field must be filled: AWS access key: Specifies the access key required for authentication. Key vault: Select the key vault from the dropdown Secret key path: The path in the key vault where your secret key is stored. The system uses this path to find and retrieve the key during authentication. Note: Replication operations will be interrupted if the session token expires. The maximum validity period for the session token is 36 hours, although the default duration is often shorter. To prevent disruptions in replication processes, it is crucial to renew the session token before it expires.
  Microsoft Azure Storage	Method: Specify the authentication method and possible values are Azure SAS token or Azure Shared Key. When selecting Azure SAS token as the authentication method, the following field must be filled: SAS token: Specify the SAS (Shared Access Signature) token required for authentication. When selecting Azure Shared Key as the authentication method, the following fields must be filled: Account name: Specifies the account name associated with the Azure storage account. Access token: Specifies the token used to authenticate and authorize access to secured storage. When selecting Key Vault as the authentication method, the following field must be filled: Account name: Specifies the account name associated with the Azure storage account. Key vault: Select the key vault from the dropdown Secret key path: The path in the key vault where your secret key is stored. The system uses this path to find and retrieve the key during authentication. Note: Replication operations will be interrupted if the SAS token expires. The maximum validity period for the SAS token is 7 days, although the default duration is often shorter. To prevent disruptions in replication processes, it is crucial to renew the session token before it expires.
  Google Cloud Storage	Service account JSON: Specifies the Google service account value, which is the unique identifier used to authenticate and authorize access to Google Cloud Platform (GCP) services and resources. When selecting Key Vault as the authentication method, the following field must be filled: Key vault: Select the key vault from the dropdown Secret key path: The path in the key vault where your secret key is stored. The system uses this path to find and retrieve the key during authentication.

• Directory: Refers to the directory in the cloud storage where the data is temporarily staged.
• Storage integration: Stores an identity and access management (IAM) entity for external cloud storage and allows settings for permitted or restricted locations, such as Amazon S3. Cloud provider administrators assign permissions for these locations, enabling seamless operations without requiring credentials for stage creation or data loading or unloading tasks.
  Note: This field is mandatory for connections using Google Cloud Storage, Microsoft Azure Storage with shared key authentication, and Amazon S3. In fact, it is mandatory regardless of the storage mechanism being used.

• Staging schema: Specifies the schema in which the staging tables should be created. This is a mandatory field.
• Duration: Allows you to select the duration limit for batch apply.
• Specify batch apply duration: Specifies the duration after which the changes are applied to Snowflake. This setting determines how frequently changes are synchronized and applied to the Snowflake database. The default value is 60 seconds.
• Batch apply duration format: Allows you to select the time format. This field is enabled when the Duration field is set to specify.
• Records: Allows you to select the size limit for batch apply operations.
• Specify the number of records: Specifies the number of records after which the changes are applied to Snowflake.
• Place value: Allows you to select the place value.

Step 3: Source data configuration

Configure source data by selecting schemas, tables, and columns for replication.

• Log reader: For IBM i data connections, the Log reader for journal dropdown list displays. Log readers may already exist. If not, you can autogenerate one. Select Configuration to open the Log Reader Configuration window, and make edits as needed.
• Schemas (Libraries): Displays all schemas available for the selected data connection. Connect to the source datastore and extract the tables for replication. Schema registry can be configured in different modes:
  • backward compatibility: When you have a version, you can only delete in the mode.
  • forward compatibility: In this mode, you can only add new rows or modify information in the Target fields.
  • full mode: In this mode, you can add, remove, and modify information in the Target fields.
• Tables
  • The Journal column shows the name of the journal associated with the tables in the schema.
  • If you are autogenerating a new log reader, the Journal column is filtered to show All Journals. You can filter the list by entering some or all of the name in the search field and selecting the Filter button. Select tables that use the same journal.
  • If you selected an existing log reader, the Tables and Journal columns are filtered to show only those items associated with the log reader.
  • Advanced table selection: Use the Advanced Search to quickly locate and select multiple tables when working with large datasets. Select Advanced Search above the Tables pane to open the filter dialog. You can enter:
    • A CSV list, comma-separated list of table names to match and select multiple tables at once. Example: CUSTOMERS, ORDERS, SALES_DATA.
    • A regular expression (RegEX) pattern to match table names dynamically. Example: ^sales_[0-9]{4}$ matches tables like sales_2020, sales_2021.
    Matching tables are automatically selected. If any entries do not match, an inline error is shown.
• Columns: The Column and Data Type columns open showing the name of each column in the table and the data type for each column (for example, DECIMAL, VARCHAR, or other types) and the content of the table with the schema applied.

Step 4: Mapping

Map source tables to target tables and configure column mappings. Depending on your source, target, and use case, refer to the relevant mapping topic:

• RRN mapping: For DB2 for IBM i sources with Google BigQuery targets, using Record Reference Numbers to track changes.
• Journal mapping: Creates an audit trail of changes for DB2 for IBM i to Google BigQuery pipelines without modifying the target dataset.
• Kafka mapping: Configure topic and subject creation for Kafka targets.
• Create target table: Create target tables in BigQuery, SQL Server, or Snowflake if they do not exist.
• Manage mapping actions: Change dataset, field, column, and table mappings; add row filters and configure soft deletes.

Step 5: Review pipline configuration

Download JCL files

When your replication project includes one or more pipelines that use a Db2 for z/OS JDBC data connection, each time you deploy the project you must first download the latest Job Control Language (JCL) files which are used to run and manage processes on z/OS systems. After you download the files, each file must be copied to and run on the z/OS server associated with the agent referenced in the file name so that the latest capture configuration is active on those servers.