### Post-Installation CLI Setup Source: https://transferwise.github.io/pipelinewise/_sources/installation_guide/installation.rst.txt Commands to activate the virtual environment and initialize the CLI after a source installation. ```bash $ make pipelinewise all_connectors (...installation usually takes 5-10 minutes...) -------------------------------------------------------------------------- PipelineWise installed successfully -------------------------------------------------------------------------- To start CLI: $ source /Users/jack/pipelinewise/.virtualenvs/pipelinewise/bin/activate $ export PIPELINEWISE_HOME=/Users/jack/pipelinewise $ pipelinewise status -------------------------------------------------------------------------- ``` -------------------------------- ### Install PipelineWise from Source Source: https://transferwise.github.io/pipelinewise/_sources/installation_guide/installation.rst.txt Clone the repository and run the make command to install the CLI and connectors into virtual environments. ```bash $ git clone https://github.com/transferwise/pipelinewise.git $ cd ./pipelinewise $ make pipelinewise all_connectors ``` -------------------------------- ### Sample tap-postgres YAML Configuration Source: https://transferwise.github.io/pipelinewise/_sources/connectors/taps/postgres.rst.txt Example structure for the tap-postgres configuration file. ```yaml --- # ------------------------------------------------------------------------------ # General Properties # ------------------------------------------------------------------------------ id: "postgres_sample" # Unique identifier of the tap name: "Sample Postgres Database" # Name of the tap type: "tap-postgres" # !! THIS SHOULD NOT CHANGE !! owner: "somebody@foo.com" # Data owner to contact #send_alert: False # Optional: Disable all configured alerts on this tap #slack_alert_channel: "#tap-channel" # Optional: Sending a copy of specific tap alerts to this slack channel ``` -------------------------------- ### Activate environment and verify installation Source: https://transferwise.github.io/pipelinewise/_sources/installation_guide/installation.rst.txt Activate the virtual environment, set the PIPELINEWISE_HOME variable, and check the status to confirm successful installation. ```bash $ source /Users/jack/pipelinewise/.virtualenvs/pipelinewise/bin/activate $ export PIPELINEWISE_HOME=/Users/jack/pipelinewise $ pipelinewise status Tap ID Tap Type Target ID Target Type Enabled Status Last Sync Last Sync Result -------- ---------- ----------- ------------- --------- -------- ----------- ------------------ 0 pipeline(s) ``` -------------------------------- ### Initialize PipelineWise environment Source: https://transferwise.github.io/pipelinewise/installation_guide/installation.html Activates the virtual environment and sets the home directory variable after installation. ```bash $ source /Users/jack/pipelinewise/.virtualenvs/pipelinewise/bin/activate $ export PIPELINEWISE_HOME=/Users/jack/pipelinewise $ pipelinewise status ``` -------------------------------- ### Configure PartialSync with sync_start_from Source: https://transferwise.github.io/pipelinewise/_sources/concept/fastsync.rst.txt Define a table replication method using a static start value for incremental synchronization. ```yaml tables: - table_name: "my_table" replication_method: "LOG_BASED" sync_start_from: column: "updated_at" static_value: "2024-01-01" # or use dynamic_value instead drop_target_table: false ``` -------------------------------- ### Install specific PipelineWise connectors Source: https://transferwise.github.io/pipelinewise/_sources/installation_guide/installation.rst.txt Use the pw_connector variable to specify a comma-separated list of connectors to install. ```bash $ make pipelinewise connectors -e pw_connector=tap-mysql,tap-postgres,target-snowflake ``` -------------------------------- ### Prepare MySQL to Snowflake replication files Source: https://transferwise.github.io/pipelinewise/_sources/installation_guide/creating_pipelines.rst.txt Renames sample configuration files to active files to begin the setup process. ```bash $ cd pipelinewise_samples $ mv tap_mysql_mariadb.yml.sample tap_mysql_sample.yml $ mv target_snowflake.yml.sample target_snowflake.yml ``` -------------------------------- ### Verify PipelineWise Installation Source: https://transferwise.github.io/pipelinewise/_sources/installation_guide/installation.rst.txt Check the status of the installation to ensure the CLI is functional. ```bash $ pipelinewise status Tap ID Tap Type Target ID Target Type Enabled Status Last Sync Last Sync Result -------- ------------ ------------ --------------- --------- -------- ----------- ------------------ 0 pipeline(s) ``` -------------------------------- ### Slack Tap YAML Configuration Source: https://transferwise.github.io/pipelinewise/_sources/connectors/taps/slack.rst.txt Example configuration file for the Slack tap, defining connection details and optional sync parameters. ```yaml --- # ------------------------------------------------------------------------------ # General Properties # ------------------------------------------------------------------------------ id: "slack" # Unique identifier of the tap name: "Slack" # Name of the tap type: "tap-slack" # !! THIS SHOULD NOT CHANGE !! owner: "somebody@foo.com" # Data owner to contact #send_alert: False # Optional: Disable all configured alerts on this tap #slack_alert_channel: "#tap-channel" # Optional: Sending a copy of specific tap alerts to this slack channel # ------------------------------------------------------------------------------ # Source (Tap) - Slack connection details # ------------------------------------------------------------------------------ db_conn: token: "" # Slack API token start_date: "2020-09-01" # Start date. Data will be synced incrementally starting from this data #channels: ["ID1", "ID2", "ID3"] # Optional: By default, the tap will sync all channels it has been invited to. # However, you can limit the tap to sync only the channels you specify by # adding their IDs to the config #exclude_archived: "false" # Optional: You can control whether or not the tap will sync archived channels # by including the following in the tap config #private_channels: "false" # Optional:, you can also specify whether you want to sync private # channels. By default private channels not synced #join_public_channels: "false" # Optional: Auto-join every public channel. # If you do not elect to have the tap join all public channels you must # invite the bot to all channels you wish to sync. #date_window_size: "5" # Optional: Due to the potentially high volume of data when syncing certain streams # (messages, files, threads) this tap implements date windowing based on # a configuration parameter. #lookback_window: 14 # Optional: Number of days to look back before the incremental start date. # This is useful to get all data from child streams that can't be extracted # incrementally, for example the message threads (conversation.replies method). # Default is 14 days. ``` -------------------------------- ### Configure Mixpanel Tap YAML Source: https://transferwise.github.io/pipelinewise/_sources/connectors/taps/mixpanel.rst.txt Example configuration file for the Mixpanel tap, defining connection details and target settings. ```yaml --- # ------------------------------------------------------------------------------ # General Properties # ------------------------------------------------------------------------------ id: "mixpanel" # Unique identifier of the tap name: "Mixpanel" # Name of the tap type: "tap-mixpanel" # !! THIS SHOULD NOT CHANGE !! owner: "somebody@foo.com" # Data owner to contact #send_alert: False # Optional: Disable all configured alerts on this tap #slack_alert_channel: "#tap-channel" # Optional: Sending a copy of specific tap alerts to this slack channel # ------------------------------------------------------------------------------ # Source (Tap) - Mixpanel connection details # ------------------------------------------------------------------------------ db_conn: api_secret: "" # Mixpanel API secret start_date: "2020-10-01" # The default value to use if no bookmark exists for an endpoint date_window_size: 30 # Number of days for date window looping through transactional endpoints # with from_date and to_date. Default date_window_size is 30 days. # Clients with large volumes of events may want to decrease this to 14, 7, # or even down to 1-2 days. attribution_window: 1 # Latency minimum number of days to look-back to account for delays in # attributing accurate results. Default attribution window is 5 days. project_timezone: "Europe/London" # Time zone in which integer date times are stored. The project timezone # may be found in the project settings in the Mixpanel console. user_agent: "tap-mixpanel " # Optional: Process and email for API logging purposes. #denest_properties: "false" # Optional: Do not denest JSON responses in `export` and `engage` streams # to avoid very wide tables. If denesting is disabled then responses are # loaded into one JSON column in the target. # Default denest_properties is false. #export_events: # Optional: List of event names to export # - event_one # - event_two # ------------------------------------------------------------------------------ # Destination (Target) - Target properties # Connection details should be in the relevant target YAML file # ------------------------------------------------------------------------------ target: "snowflake" # ID of the target connector where the data will be loaded batch_size_rows: 20000 # Batch size for the stream to optimise load performance stream_buffer_size: 0 # In-memory buffer size (MB) between taps and targets for asynchronous data pipes default_target_schema: "mixpanel" # Target schema where the data will be loaded #default_target_schema_select_permission: # Optional: Grant SELECT on schema and tables that created # - grp_power #batch_wait_limit_seconds: 3600 # Optional: Maximum time to wait for `batch_size_rows`. Available only for snowflake target. ``` -------------------------------- ### Twilio Tap Configuration YAML Source: https://transferwise.github.io/pipelinewise/_sources/connectors/taps/twilio.rst.txt Example configuration file for the Twilio tap, including required credentials and schema definitions. ```yaml id: "twilio" name: "Twilio" type: "tap-twilio" owner: "data-team@example.com" db_conn: account_sid: "" auth_token: "" start_date: "2024-01-01T00:00:00Z" user_agent: "data-team@example.com" target: "snowflake" batch_size_rows: 20000 stream_buffer_size: 0 default_target_schema: "twilio" schemas: - source_schema: "twilio" target_schema: "twilio" tables: - table_name: "workspaces" - table_name: "activities" - table_name: "events" - table_name: "tasks" ``` -------------------------------- ### MySQL Tap YAML Configuration Source: https://transferwise.github.io/pipelinewise/connectors/taps/mysql.html Example configuration file for tap-mysql, defining connection details, GTID settings, and target destination parameters. ```yaml --- # ------------------------------------------------------------------------------ # General Properties # ------------------------------------------------------------------------------ id: "mysql_sample" # Unique identifier of the tap name: "Sample MySQL Database" # Name of the tap type: "tap-mysql" # !! THIS SHOULD NOT CHANGE !! owner: "somebody@foo.com" # Data owner to contact #send_alert: False # Optional: Disable all configured alerts on this tap #slack_alert_channel: "#tap-channel" # Optional: Sending a copy of specific tap alerts to this slack channel # ------------------------------------------------------------------------------ # Source (Tap) - MySQL/ MariaDB connection details # ------------------------------------------------------------------------------ db_conn: host: "" # MySQL/ MariaDB host port: 3306 # MySQL/ MariaDB port user: "" # MySQL/ MariaDB user password: "" # Plain string or vault encrypted dbname: "" # MySQL/ MariaDB database name use_gtid: true # Use GTID as the state bookmark for log-based tables engine: "mysql" # One of "mysql" or "mariadb"; used with use_gtid #replica_host: "" # Optional: MySQL/ MariaDB replica host to offload initial/FastSync # to a read replica, switch back to primary for log-based replication. # Used to resync large tables without impacting primary DB performance. #filter_dbs: "schema1,schema2" # Optional: Scan only the required schemas # to improve the performance of # data extraction #export_batch_rows: 50000 # Optional: Number of rows to export from MySQL # in one batch. Default is 50000. #session_sqls: # Optional: Run SQLs to set session variables # - SET @@session.time_zone="+0:00" # when the connection made # - SET @@session.wait_timeout=28800 # Defaults to the values listed here # - SET @@session.net_read_timeout=3600 # - SET @@session.innodb_lock_wait_timeout=3600 #fastsync_parallelism: 4 # Optional: size of multiprocessing pool used by FastSync # Min: 1 # Default: number of CPU cores # ------------------------------------------------------------------------------ # Destination (Target) - Target properties # Connection details should be in the relevant target YAML file # ------------------------------------------------------------------------------ target: "snowflake" # ID of the target connector where the data will be loaded batch_size_rows: 20000 # Batch size for the stream to optimise load performance stream_buffer_size: 0 # In-memory buffer size (MB) between taps and targets for asynchronous data pipes #batch_wait_limit_seconds: 3600 # Optional: Maximum time to wait for `batch_size_rows`. Available only for snowflake target. ``` -------------------------------- ### Snowflake Target YAML Configuration Source: https://transferwise.github.io/pipelinewise/_sources/connectors/targets/snowflake.rst.txt Example YAML configuration file for defining the Snowflake target connection details. ```yaml --- # ------------------------------------------------------------------------------ # General Properties # ------------------------------------------------------------------------------ id: "snowflake" # Unique identifier of the target name: "Snowflake" # Name of the target type: "target-snowflake" # !! THIS SHOULD NOT CHANGE !! # ------------------------------------------------------------------------------ # Target - Data Warehouse connection details # ------------------------------------------------------------------------------ db_conn: account: "rtxxxxx.eu-central-1" # Snowflake account dbname: "" # Snowflake database name user: "" # Snowflake user private_key: "" # File contains PEM format for connecting to Snowflake warehouse: "" # Snowflake virtual warehouse iceberg_create: false # Create new tables as Iceberg tables (only available for pure Singer replications) # We use an external stage on S3 to load data into Snowflake # S3 Profile based authentication aws_profile: "" # AWS profile name, if not provided, the AWS_PROFILE environment # variable or the 'default' profile will be used, if not # available, then IAM role attached to the host will be used. ``` -------------------------------- ### Configure Unix Cron for Pipelines Source: https://transferwise.github.io/pipelinewise/_sources/user_guide/scheduling.rst.txt Example crontab entries to schedule individual pipeline syncs at varying intervals. ```bash */5 * * * * pipelinewise run_tap --tap microserv_1 --target snowflake # Sync every 5 minutes 0 * * * * pipelinewise run_tap --tap microserv_2 --target snowflake # Sync every hour 0 */3 * * * pipelinewise run_tap --tap microserv_3 --target snowflake # Sync every three hours 0 0 * * * pipelinewise run_tap --tap microserv_4 --target snowflake # Sync every midnight 0 0 * * 6 pipelinewise run_tap --tap microserv_5 --target snowflake # Sync every Saturday ``` -------------------------------- ### Salesforce Tap Configuration YAML Source: https://transferwise.github.io/pipelinewise/_sources/connectors/taps/salesforce.rst.txt Example configuration file for the Salesforce tap, specifying authentication credentials, API type, and target schema. ```yaml id: "salesforce" name: "Salesforce" type: "tap-salesforce" owner: "data-team@example.com" db_conn: client_id: "" client_secret: "" refresh_token: "" start_date: "2024-01-01T00:00:00Z" api_type: "BULK" target: "snowflake" batch_size_rows: 20000 stream_buffer_size: 0 default_target_schema: "salesforce" schemas: - source_schema: "salesforce" target_schema: "salesforce" tables: - table_name: "Account" - table_name: "Contact" - table_name: "Opportunity" ``` -------------------------------- ### Configure Metadata and Delete Settings in Tap YAML Source: https://transferwise.github.io/pipelinewise/_sources/user_guide/metadata_columns.rst.txt Example configuration for a MySQL tap enabling metadata columns and disabling hard deletes. ```yaml id: "mysql_orders" name: "MySQL orders" type: "tap-mysql" add_metadata_columns: true hard_delete: false db_conn: # ... ``` -------------------------------- ### Missing Binlog Error Log Source: https://transferwise.github.io/pipelinewise/_sources/user_guide/troubleshooting.rst.txt Example log output when the required binary log file has been purged from the source server. ```text Exception: Unable to replicate binlog stream because the following binary log(s) no longer exist: mysql-bin.000119 ``` -------------------------------- ### View sample project structure Source: https://transferwise.github.io/pipelinewise/_sources/installation_guide/creating_pipelines.rst.txt Displays the directory structure generated by the init command, including various tap and target samples. ```bash └── pipelinewise_samples ├── README.md ├── config.yml ├── tap_jira.yml.sample ├── tap_kafka.yml.sample ├── tap_github.yml.sample ├── tap_google_analytics.yml.sample ├── tap_mixpanel.yml.sample ├── tap_mongodb.yml.sample ├── tap_mysql_mariadb.yml.sample ├── tap_oracle.yml.sample ├── tap_postgres.yml.sample ├── tap_s3_csv.yml.sample ├── tap_salesforce.yml.sample ├── tap_shopify.yml.sample ├── tap_slack.yml.sample ├── tap_snowflake.yml.sample ├── tap_twilio.yml.sample ├── tap_zendesk.yml.sample ├── tap_zuora.yml.sample ├── target_postgres.yml.sample ├── target_s3_csv.yml.sample └── target_snowflake.yml.sample ``` -------------------------------- ### Initialize a PipelineWise project Source: https://transferwise.github.io/pipelinewise/_sources/installation_guide/creating_pipelines.rst.txt Creates a new directory containing sample YAML configuration files for various connectors. ```bash $ pipelinewise init --name pipelinewise_samples ``` -------------------------------- ### MySQL Connection Error Log Source: https://transferwise.github.io/pipelinewise/_sources/user_guide/troubleshooting.rst.txt Example log output for a lost connection error during a MySQL query. ```text pymysql.err.OperationalError: (2013, 'Lost connection to MySQL server during query ([Errno 104] Connection reset by peer)') ``` -------------------------------- ### Data Diff Failure Alert Output Source: https://transferwise.github.io/pipelinewise/_sources/user_guide/data_diff.rst.txt Example of the output format generated when a data-diff check fails. ```text data-diff FAIL snowflake/payments/public/transfers window 2026-07-29T10:00:00+00:00 → 2026-07-29T11:00:00+00:00 run_id 2bd3e725-38fc-48c1-b565-b4f20e5bc7dd ``` -------------------------------- ### Use Encrypted Password in YAML Source: https://transferwise.github.io/pipelinewise/_sources/user_guide/encrypting_passwords.rst.txt Example of integrating the encrypted vault string into a database connection configuration. ```bash db_conn: host: "mysql_source_database" port: 3306 user: "jack_replica" password: !vault | $ANSIBLE_VAULT;1.1;AES256 66633736626365386334633463383431353762623562663733623833626637383762343430626163 3330383435323737383766333639616138363235356135360a343735366137393763323636353739 63646266353131363436363965336561323336663032336334323236616237363430613432386335 6631613665336365640a643531363765663631306431363433623536363061316234643737323465 34626363656166373230303162623531643638656665633731333338333464633565 dbname: "my_database" ``` -------------------------------- ### Import Pipeline Configuration Source: https://transferwise.github.io/pipelinewise/_sources/installation_guide/creating_pipelines.rst.txt Command to import YAML configurations and initialize the pipeline environment. ```bash $ pipelinewise import_config --dir pipelinewise_samples ... detailed messages about import and discovery... ------------------------------------------------------- IMPORTING YAML CONFIGS FINISHED ------------------------------------------------------- Total targets to import : 1 Total taps to import : 1 Taps imported successfully : 1 Taps failed to import : [] Runtime : 0:00:01.835720 ------------------------------------------------------- ``` -------------------------------- ### FastSync Table Not Found Error Log Source: https://transferwise.github.io/pipelinewise/_sources/user_guide/troubleshooting.rst.txt Example of the critical log message generated when a table cannot be located in the source database. ```text CRITICAL: {SCHEMA_NAME}.TABLE_NAME table not found. ``` -------------------------------- ### Import PipelineWise configuration Source: https://transferwise.github.io/pipelinewise/installation_guide/creating_pipelines.html Use this command to import YAML configurations from a specified directory and trigger database discovery. ```bash $ pipelinewise import_config --dir pipelinewise_samples ``` -------------------------------- ### Define Target and Performance Settings Source: https://transferwise.github.io/pipelinewise/_sources/connectors/taps/kafka.rst.txt Settings for the destination target, batch sizes, and schema permissions. ```yaml # ------------------------------------------------------------------------------ # Destination (Target) - Target properties # Connection details should be in the relevant target YAML file # ------------------------------------------------------------------------------ target: "snowflake" # ID of the target connector where the data will be loaded batch_size_rows: 20000 # Batch size for the stream to optimise load performance stream_buffer_size: 0 # In-memory buffer size (MB) between taps and targets for asynchronous data pipes default_target_schema: "kafka" # Target schema where the data will be loaded default_target_schema_select_permission: # Optional: Grant SELECT on schema and tables that created - grp_stats #batch_wait_limit_seconds: 3600 # Optional: Maximum time to wait for `batch_size_rows`. Available only for snowflake target. # Options only for Snowflake target #archive_load_files: False # Optional: when enabled, the files loaded to Snowflake will also be stored in `archive_load_files_s3_bucket` #archive_load_files_s3_prefix: "archive" # Optional: When `archive_load_files` is enabled, the archived files will be placed in the archive S3 bucket under this prefix. #archive_load_files_s3_bucket: "" # Optional: When `archive_load_files` is enabled, the archived files will be placed in this bucket. (Default: the value of `s3_bucket` in target snowflake YAML) ``` -------------------------------- ### Configure PostgreSQL for Logical Replication Source: https://transferwise.github.io/pipelinewise/_sources/connectors/taps/postgres.rst.txt Required parameters for the postgresql.conf file to enable log-based replication. ```ini wal_level=logical max_replication_slots=5 max_wal_senders=5 ``` -------------------------------- ### Configure MySQL Table Replication Source: https://transferwise.github.io/pipelinewise/_sources/connectors/taps/mysql.rst.txt Defines table replication methods and optional sync start parameters for MySQL sources. ```yaml # OPTIONAL: Load time transformations #transformations: # - column: "last_name" # Column to transform # type: "SET-NULL" # Transformation type # You can add as many tables as you need... - table_name: "table_two" replication_method: "LOG_BASED" # Important! Log based must be enabled in MySQL - table_name: "table_three" replication_method: "LOG_BASED" sync_start_from: # Optional, applies for the first sync and fast sync column: "column_name" # Column name to be picked for partial sync with incremental or timestamp value static_value: "start_value" # A static value which the first sync always starts from column >= static_value drop_target_table: true # Optional, drops target table before syncing. default value is false - table_name: "table_four" replication_method: "LOG_BASED" sync_start_from: # Optional, applies for the first sync and fast sync column: "column_name" # Column name to be picked for partial sync with incremental or timestamp value # The query must return exactly one row with one column. The first sync starts # from rows where column >= the returned value. dynamic_value: "SELECT DATE_SUB(MAX(updated_at), INTERVAL 7 DAY) FROM table_four" drop_target_table: true # Optional, drops target table before syncing. default value is false # You can add as many schemas as you need... # Uncomment this if you want to replicate tables from multiple schemas #- source_schema: "another_schema_in_mysql" # target_schema: "another_schema_in_target" # Note: static and dynamic values cannot be defined together for a table; only one can be used. ``` -------------------------------- ### Create Snowflake Role and Permissions Source: https://transferwise.github.io/pipelinewise/_sources/connectors/targets/snowflake.rst.txt Sets up a dedicated role with the necessary privileges to access the database, schema, stage, file format, and warehouse. ```bash CREATE OR REPLACE ROLE ppw_target_snowflake; GRANT USAGE ON DATABASE {database} TO ROLE ppw_target_snowflake; GRANT CREATE SCHEMA ON DATABASE {database} TO ROLE ppw_target_snowflake; GRANT USAGE ON SCHEMA {database}.{schema} TO role ppw_target_snowflake; GRANT USAGE ON STAGE {database}.{schema}.{stage_name} TO ROLE ppw_target_snowflake; GRANT USAGE ON FILE FORMAT {database}.{schema}.{file_format_name} TO ROLE ppw_target_snowflake; GRANT USAGE ON WAREHOUSE {warehouse} TO ROLE ppw_target_snowflake; ``` -------------------------------- ### Configure MySQL server settings Source: https://transferwise.github.io/pipelinewise/_sources/connectors/taps/mysql.rst.txt Add these settings to the [mysqld] section of the my.cnf file to enable required binary logging features. ```ini [mysqld] binlog_format=ROW binlog_row_image=FULL expire_logs_days=7 log_bin=mysql-binlog log_slave_updates=1 ``` -------------------------------- ### Sample YAML configuration for tap-mongodb Source: https://transferwise.github.io/pipelinewise/_sources/connectors/taps/mongodb.rst.txt Provides a template for the tap-mongodb configuration file, including database connection details and replication settings. ```yaml --- id: "tap_mongo" # Unique identifier of the tap name: "MongoDB tap" # Human-readable name of the tap type: "tap-mongodb" # Connector type; do not change owner: "foo@bar.com" # Data owner to contact #send_alert: false # Optional: Disable configured alerts for this tap #slack_alert_channel: "#tap-channel" # Optional: Send a copy of alerts to this Slack channel db_conn: host: "mongodb_host1,mongodb_host2,mongodb_host3" # Comma-separated MongoDB hosts port: 27017 # MongoDB port srv: "false" # Use "true" for MongoDB Atlas; port is then ignored user: "PipelineWiseUser" # User with permission to read the source collections password: "" # Plain string or Vault encrypted auth_database: "admin" # Database against which the user authenticates dbname: "my_db" # MongoDB database to replicate replica_set: "my_replica_set" # Optional: Replica set name; default is null #write_batch_rows: 50000 # Optional: Rows written to a CSV batch; default 50000 #update_buffer_size: 1 # Optional, LOG_BASED: Buffered update operations; default 1 #await_time_ms: 1000 # Optional, LOG_BASED: Wait for changes before exit; default 1000 ms #fastsync_parallelism: 4 # Optional: FastSync process pool size; defaults to CPU count target: "my_target" # ID of the target connector batch_size_rows: 1000 # Rows sent to the target in each batch stream_buffer_size: 0 # In-memory tap-to-target buffer size in MB default_target_schema: "my_db" # Optional: Default destination schema #default_target_schema_select_permissions: # Optional: Groups granted SELECT access # - grp_power #batch_wait_limit_seconds: 3600 # Optional, Snowflake: Flush a partial batch after this time ``` -------------------------------- ### Configure Kafka Tap Protobuf Support Source: https://transferwise.github.io/pipelinewise/_sources/connectors/taps/kafka.rst.txt Experimental settings for enabling Protobuf message format and defining the schema. ```yaml # -------------------------------------------------------------------------- # Protobuf support - Experimental # -------------------------------------------------------------------------- #message_format: protobuf # (Default: json) Supported message formats are json and protobuf. #proto_schema: | # Protobuf message format in .proto syntax. Required if the message_format is protobuf. # syntax = "proto3"; # # message ProtoMessage { # string query = 1; # int32 page_number = 2; # int32 result_per_page = 3; # } #proto_classes_dir: # (Default: current working dir) Directory where runtime compiled proto classes are stored #debug_contexts: "" # comma separated list of debug contexts to enable for the consumer [see librkafka](https://github.com/confluentinc/librdkafka/blob/master/INTRODUCTION.md#debug-contexts) ``` -------------------------------- ### Postgres Tap Configuration YAML Source: https://transferwise.github.io/pipelinewise/_sources/connectors/taps/postgres.rst.txt Defines the source and target schemas, table replication methods, and optional sync start parameters for the Postgres tap. ```yaml - source_schema: "public" # Source schema in postgres with tables target_schema: "repl_pg_public" # Target schema in the destination Data Warehouse target_schema_select_permissions: # Optional: Grant SELECT on schema and tables that created - grp_stats # List of tables to replicate from Postgres to destination Data Warehouse # # Please check the Replication Strategies section in the documentation to understand the differences. # For LOG_BASED replication method you might need to adjust the source PostgreSQL configuration. tables: - table_name: "table_one" replication_method: "INCREMENTAL" # One of INCREMENTAL, LOG_BASED and FULL_TABLE replication_key: "last_update" # Important: Incremental load always needs replication key # OPTIONAL: Load time transformations #transformations: # - column: "last_name" # Column to transform # type: "SET-NULL" # Transformation type # You can add as many tables as you need... - table_name: "table_two" replication_method: "LOG_BASED" # Important! Log based must be enabled in PostgreSQL - table_name: "table_three" replication_method: "LOG_BASED" sync_start_from: # Optional, applies for the first sync and fast sync column: "column_name" # column name to be picked for partial sync with incremental or timestamp value static_value: "start_value" # A static value which the first sync always starts from column >= static_value drop_target_table: true # Optional, drops target table before syncing. default value is false - table_name: "table_four" replication_method: "LOG_BASED" sync_start_from: # Optional, applies for the first sync and fast sync column: "column_name" # Column name to be picked for partial sync with incremental or timestamp value dynamic_value: "SELECT MAX(updated_at) - INTERVAL '7 days' FROM table_four" drop_target_table: true # Optional, drops target table before syncing. default value is false # You can add as many schemas as you need... # Uncomment this if you want to replicate tables from multiple schemas #- source_schema: "another_schema_in_postgres" # target_schema: "another_schema_in_target" # Note: static and dynamic values cannot be defined together for a table; only one can be used. ``` -------------------------------- ### Configure tap-kafka YAML Source: https://transferwise.github.io/pipelinewise/_sources/connectors/taps/kafka.rst.txt A sample YAML configuration file for the Kafka tap, defining connection details, primary keys, and consumer settings. ```yaml --- # ------------------------------------------------------------------------------ # General Properties # ------------------------------------------------------------------------------ id: "kafka" # Unique identifier of the tap name: "Kafka Topic with sample data" # Name of the tap type: "tap-kafka" # !! THIS SHOULD NOT CHANGE !! owner: "somebody@foo.com" # Data owner to contact #send_alert: False # Optional: Disable all configured alerts on this tap #slack_alert_channel: "#tap-channel" # Optional: Sending a copy of specific tap alerts to this slack channel # ------------------------------------------------------------------------------ # Source (Tap) - Kafka connection details # ------------------------------------------------------------------------------ db_conn: group_id: "myGroupId" bootstrap_servers: "kafka1.foo.com:9092,kafka2.foo.com:9092,kafka3.foo.com:9092" topic: "myKafkaTopic" # -------------------------------------------------------------------------- # Optionally you can define primary key(s) from the kafka JSON messages. # If primary keys defined then extra column(s) will be added to the output # singer stream with the extracted values by /slashed/paths ala XPath selectors. # -------------------------------------------------------------------------- primary_keys: transfer_id: "/transferMetadata/transferId" # -------------------------------------------------------------------------- # Additionally you can specify whether to use Kafka message key as a primary key. # If custom primary keys were specified earlier, message key property is ignored # and custom PKs are used instead. # ! Message key should be a valid utf-8 encoded string. # -------------------------------------------------------------------------- #use_message_key: true # Optional. Use the Kafka message key as the primary key. # Default: true. Ignored when primary_keys is configured. #initial_start_time: # (Default: latest) Start time reference of the message consumption if # no bookmarked position in state.json. One of: latest, earliest or an # ISO-8601 formatted timestamp string. # -------------------------------------------------------------------------- # Kafka Consumer optional parameters. Commented values are default values. # -------------------------------------------------------------------------- #max_runtime_ms: 300000 # The maximum time for the tap to collect new messages from Kafka topic. #consumer_timeout_ms: 10000 # KafkaConsumer setting. Number of milliseconds to block during message iteration before raising StopIteration #session_timeout_ms: 30000 # KafkaConsumer setting. The timeout used to detect failures when using Kafka’s group management facilities. #heartbeat_interval_ms: 10000 # KafkaConsumer setting. The expected time in milliseconds between heartbeats to the consumer coordinator when using Kafka’s group management facilities. #max_poll_interval_ms: 300000 # KafkaConsumer setting. The maximum delay between invocations of poll() when using consumer group management. #commit_interval_ms: 5000 # Number of milliseconds between two commits. This is different than the kafka auto commit feature. Tap-kafka sends commit messages automatically but only when the data consumed successfully and persisted to local store. ``` -------------------------------- ### Postgres Target YAML Configuration Source: https://transferwise.github.io/pipelinewise/_sources/connectors/targets/postgres.rst.txt A sample YAML configuration file for the target-postgres connector, defining connection details and target properties. ```yaml --- # ------------------------------------------------------------------------------ # General Properties # ------------------------------------------------------------------------------ id: "postgres_dwh" # Unique identifier of the target name: "Postgres Data Warehouse" # Name of the target type: "target-postgres" # !! THIS SHOULD NOT CHANGE !! # ------------------------------------------------------------------------------ # Target - Data Warehouse connection details # ------------------------------------------------------------------------------ db_conn: host: "" # PostgreSQL host port: 5432 # PostgreSQL port user: "" # PostgreSQL user password: "" # Plain string or vault encrypted dbname: "" # PostgreSQL database name #ssl: "true" # Optional: Using SSL via postgres sslmode 'require' option. # If the server does not accept SSL connections or the client # certificate is not recognized the connection will fail ``` -------------------------------- ### List PipelineWise Log Files Source: https://transferwise.github.io/pipelinewise/_sources/user_guide/logging.rst.txt Displays the contents of the log directory for a specific target and tap to verify generated log files. ```bash $ ls -lah ~/.pipelinewise/snowflake/fx/log/ -rw-rw-r-- 1 pipelinewise pipelinewise 13K May 8 00:02 snowflake-fx-20190508_000038.singer.log.success -rw-rw-r-- 1 pipelinewise pipelinewise 13K May 8 00:31 snowflake-fx-20190508_003014.singer.log.success -rw-rw-r-- 1 pipelinewise pipelinewise 13K May 8 01:04 snowflake-fx-20190508_010031.singer.log.success -rw-rw-r-- 1 pipelinewise pipelinewise 13K May 8 01:33 snowflake-fx-20190508_013036.singer.log.success -rw-rw-r-- 1 pipelinewise pipelinewise 13K May 8 02:02 snowflake-fx-20190508_020037.singer.log.success -rw-rw-r-- 1 pipelinewise pipelinewise 13K May 8 02:32 snowflake-fx-20190508_023029.singer.log.success -rw-rw-r-- 1 pipelinewise pipelinewise 13K May 8 03:02 snowflake-fx-20190508_030032.singer.log.success -rw-rw-r-- 1 pipelinewise pipelinewise 13K May 8 03:31 snowflake-fx-20190508_033031.singer.log.success -rw-rw-r-- 1 pipelinewise pipelinewise 13K May 8 04:02 snowflake-fx-20190508_040033.fastsync.log.success -rw-rw-r-- 1 pipelinewise pipelinewise 13K May 8 04:04 snowflake-fx-20190508_040222.singer.log.success -rw-rw-r-- 1 pipelinewise pipelinewise 13K May 8 04:30 snowflake-fx-20190508_043015.singer.log.success -rw-rw-r-- 1 pipelinewise pipelinewise 13K May 8 05:01 snowflake-fx-20190508_050035.singer.log.success -rw-rw-r-- 1 pipelinewise pipelinewise 13K May 8 05:31 snowflake-fx-20190508_053030.singer.log.failed -rw-rw-r-- 1 pipelinewise pipelinewise 13K May 8 06:02 snowflake-fx-20190508_060037.singer.log.success -rw-rw-r-- 1 pipelinewise pipelinewise 13K May 8 06:33 snowflake-fx-20190508_063032.singer.log.success -rw-rw-r-- 1 pipelinewise pipelinewise 8K May 8 07:03 snowflake-fx-20190508_070036.singer.log.running ``` -------------------------------- ### Build PipelineWise Docker Image Source: https://transferwise.github.io/pipelinewise/_sources/installation_guide/installation.rst.txt Clone the repository and build the Docker image to isolate dependencies. ```bash $ git clone https://github.com/transferwise/pipelinewise.git $ cd ./pipelinewise $ docker build -t pipelinewise:latest . ``` -------------------------------- ### Monitor Sync Progress Source: https://transferwise.github.io/pipelinewise/_sources/user_guide/troubleshooting.rst.txt Commands to follow live sync logs or inspect temporary files generated during the synchronization process. ```bash tail -f ~/.pipelinewise///log/*running ``` ```bash ls -lah ~/.pipelinewise/tmp/** ``` -------------------------------- ### Create PipelineWise Database User Source: https://transferwise.github.io/pipelinewise/_sources/connectors/taps/postgres.rst.txt SQL commands to create a dedicated user and grant necessary read permissions for replication. ```sql CREATE USER pipelinewise WITH ENCRYPTED PASSWORD '' GRANT CONNECT ON DATABASE TO pipelinewise GRANT USAGE ON SCHEMA TO pipelinewise GRANT SELECT ON ALL TABLES IN SCHEMA TO pipelinewise ``` -------------------------------- ### Create Vault Password File Source: https://transferwise.github.io/pipelinewise/_sources/user_guide/encrypting_passwords.rst.txt Sets up a restricted-permission file to store the master vault password. ```bash $ umask 077 $ printf '%s\n' 'M@st3rP@ssw0rd' > vault-password.txt ```