PipelineWise

https://transferwise.github.io/pipelinewise/
Admin
PipelineWise is a data pipeline framework using the Singer.io specification to ingest and replicate...
Tokens:49,605
Snippets:573
Trust Score:9.7
Update:1 month ago
Show doc for...
Tokens:
Raw
### 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_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>"       # 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 <api@foo.com>"  # 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: "<TWILIO_ACCOUNT_SID>"
  auth_token: "<TWILIO_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: "<HOST>"                       # MySQL/ MariaDB host
  port: 3306                           # MySQL/ MariaDB port
  user: "<USER>"                       # MySQL/ MariaDB user
  password: "<PASSWORD>"               # Plain string or vault encrypted
  dbname: "<DB_NAME>"                  # 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: "<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: "<DB_NAME>"                           # Snowflake database name
  user: "<USER>"                                # Snowflake user
  private_key: "<private_key_path>"             # File contains PEM format for connecting to Snowflake
  warehouse: "<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>"                  # 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_ID>"
  client_secret: "<CLIENT_SECRET>"
  refresh_token: "<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: "<BUCKET_NAME>"  # 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: "<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: "<HOST>"                       # PostgreSQL host
  port: 5432                           # PostgreSQL port
  user: "<USER>"                       # PostgreSQL user
  password: "<PASSWORD>"               # Plain string or vault encrypted
  dbname: "<DB_NAME>"                  # 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
```

--------------------------------

### 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 .
```

--------------------------------

### 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
```

--------------------------------

### 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/<target_id>/<tap_id>/log/*running
```

```bash
ls -lah ~/.pipelinewise/tmp/*<tap_id>*
```

--------------------------------

### 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 '<password>'
GRANT CONNECT ON DATABASE <database_name> TO pipelinewise
GRANT USAGE ON SCHEMA <schema_name> TO pipelinewise
GRANT SELECT ON ALL TABLES IN SCHEMA <schema_name> 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
```
Tokens:
Raw