Fivetran

Important Capabilities

Capability	Status	Notes
Platform Instance	✅	Enabled by default

This plugin extracts fivetran users, connectors, destinations and sync history. This plugin is in beta and has only been tested on Snowflake connector.

CLI based Ingestion

Install the Plugin

pip install 'acryl-datahub[fivetran]'

Config Details

Options

Note that a . is used to denote nested fields in the YAML recipe.

Field	Description
fivetran_log_config ✅ FivetranLogConfig	Fivetran log connector destination server configurations.
fivetran_log_config.destination_platform string	The destination platform where fivetran connector log tables are dumped. Default: snowflake
fivetran_log_config.destination_config DestinationConfig	If destination platform is 'snowflake', provide snowflake configuration.
fivetran_log_config.destination_config.account_id ❓ string	Snowflake account identifier. e.g. xy12345, xy12345.us-east-2.aws, xy12345.us-central1.gcp, xy12345.central-us.azure, xy12345.us-west-2.privatelink. Refer Account Identifiers for more details.
fivetran_log_config.destination_config.database ❓ string	The fivetran connector log database.
fivetran_log_config.destination_config.log_schema ❓ string	The fivetran connector log schema.
fivetran_log_config.destination_config.authentication_type string	The type of authenticator to use when connecting to Snowflake. Supports "DEFAULT_AUTHENTICATOR", "OAUTH_AUTHENTICATOR", "EXTERNAL_BROWSER_AUTHENTICATOR" and "KEY_PAIR_AUTHENTICATOR". Default: DEFAULT_AUTHENTICATOR
fivetran_log_config.destination_config.bucket_duration Enum	Size of the time window to aggregate usage stats. Default: DAY
fivetran_log_config.destination_config.connect_args object	Connect args to pass to Snowflake SqlAlchemy driver
fivetran_log_config.destination_config.end_time string(date-time)	Latest date of lineage/usage to consider. Default: Current time in UTC
fivetran_log_config.destination_config.options object	Any options specified here will be passed to SQLAlchemy.create_engine as kwargs.
fivetran_log_config.destination_config.password string(password)	Snowflake password.
fivetran_log_config.destination_config.private_key string	Private key in a form of '-----BEGIN PRIVATE KEY-----\nprivate-key\n-----END PRIVATE KEY-----\n' if using key pair authentication. Encrypted version of private key will be in a form of '-----BEGIN ENCRYPTED PRIVATE KEY-----\nencrypted-private-key\n-----END ECNCRYPTED PRIVATE KEY-----\n' See: https://docs.snowflake.com/en/user-guide/key-pair-auth.html
fivetran_log_config.destination_config.private_key_password string(password)	Password for your private key. Required if using key pair authentication with encrypted private key.
fivetran_log_config.destination_config.private_key_path string	The path to the private key if using key pair authentication. Ignored if private_key is set. See: https://docs.snowflake.com/en/user-guide/key-pair-auth.html
fivetran_log_config.destination_config.role string	Snowflake role.
fivetran_log_config.destination_config.scheme string	Default: snowflake
fivetran_log_config.destination_config.start_time string(date-time)	Earliest date of lineage/usage to consider. Default: Last full day in UTC (or hour, depending on bucket_duration). You can also specify relative time with respect to end_time such as '-7 days' Or '-7d'.
fivetran_log_config.destination_config.username string	Snowflake username.
fivetran_log_config.destination_config.warehouse string	Snowflake warehouse.
fivetran_log_config.destination_config.oauth_config OAuthConfiguration	oauth configuration - https://docs.snowflake.com/en/user-guide/python-connector-example.html#connecting-with-oauth
fivetran_log_config.destination_config.oauth_config.authority_url ❓ string	Authority url of your identity provider
fivetran_log_config.destination_config.oauth_config.client_id ❓ string	client id of your registered application
fivetran_log_config.destination_config.oauth_config.provider ❓ Enum	Identity provider for oauth.Supported providers are microsoft and okta.
fivetran_log_config.destination_config.oauth_config.client_secret string(password)	client secret of the application if use_certificate = false
fivetran_log_config.destination_config.oauth_config.encoded_oauth_private_key string	base64 encoded private key content if use_certificate = true
fivetran_log_config.destination_config.oauth_config.encoded_oauth_public_key string	base64 encoded certificate content if use_certificate = true
fivetran_log_config.destination_config.oauth_config.scopes array(string)
fivetran_log_config.destination_config.oauth_config.use_certificate boolean	Do you want to use certificate and private key to authenticate using oauth Default: False
platform_instance string	The instance of the platform that all assets produced by this recipe belong to
env string	The environment that all assets produced by this connector belong to Default: PROD
connector_patterns AllowDenyPattern	Regex patterns for connectors to filter in ingestion. Default: {'allow': ['.*'], 'deny': [], 'ignoreCase': True}
connector_patterns.allow array(string)
connector_patterns.deny array(string)
connector_patterns.ignoreCase boolean	Whether to ignore case sensitivity during pattern matching. Default: True
destination_to_platform_instance map(str,PlatformDetail)
destination_to_platform_instance.key.platform_instance string	The instance of the platform that all assets produced by this recipe belong to
destination_to_platform_instance.key.env string	The environment that all assets produced by DataHub platform ingestion source belong to Default: PROD
sources_to_platform_instance map(str,PlatformDetail)
sources_to_platform_instance.key.platform_instance string	The instance of the platform that all assets produced by this recipe belong to
sources_to_platform_instance.key.env string	The environment that all assets produced by DataHub platform ingestion source belong to Default: PROD
stateful_ingestion StatefulStaleMetadataRemovalConfig	Airbyte Stateful Ingestion Config.
stateful_ingestion.enabled boolean	The type of the ingestion state provider registered with datahub. Default: False
stateful_ingestion.remove_stale_metadata boolean	Soft-deletes the entities present in the last successful run but missing in the current run with stateful_ingestion enabled. Default: True

Schema

The JSONSchema for this configuration is inlined below.

{
  "title": "FivetranSourceConfig",
  "description": "Base configuration class for stateful ingestion for source configs to inherit from.",
  "type": "object",
  "properties": {
    "env": {
      "title": "Env",
      "description": "The environment that all assets produced by this connector belong to",
      "default": "PROD",
      "type": "string"
    },
    "platform_instance": {
      "title": "Platform Instance",
      "description": "The instance of the platform that all assets produced by this recipe belong to",
      "type": "string"
    },
    "stateful_ingestion": {
      "title": "Stateful Ingestion",
      "description": "Airbyte Stateful Ingestion Config.",
      "allOf": [
        {
          "$ref": "#/definitions/StatefulStaleMetadataRemovalConfig"
        }
      ]
    },
    "fivetran_log_config": {
      "title": "Fivetran Log Config",
      "description": "Fivetran log connector destination server configurations.",
      "allOf": [
        {
          "$ref": "#/definitions/FivetranLogConfig"
        }
      ]
    },
    "connector_patterns": {
      "title": "Connector Patterns",
      "description": "Regex patterns for connectors to filter in ingestion.",
      "default": {
        "allow": [
          ".*"
        ],
        "deny": [],
        "ignoreCase": true
      },
      "allOf": [
        {
          "$ref": "#/definitions/AllowDenyPattern"
        }
      ]
    },
    "sources_to_platform_instance": {
      "title": "Sources To Platform Instance",
      "description": "A mapping of the connector's all sources dataset to platform instance. Use connector id as key.",
      "default": {},
      "type": "object",
      "additionalProperties": {
        "$ref": "#/definitions/PlatformDetail"
      }
    },
    "destination_to_platform_instance": {
      "title": "Destination To Platform Instance",
      "description": "A mapping of destination dataset to platform instance. Use destination id as key.",
      "default": {},
      "type": "object",
      "additionalProperties": {
        "$ref": "#/definitions/PlatformDetail"
      }
    }
  },
  "required": [
    "fivetran_log_config"
  ],
  "additionalProperties": false,
  "definitions": {
    "DynamicTypedStateProviderConfig": {
      "title": "DynamicTypedStateProviderConfig",
      "type": "object",
      "properties": {
        "type": {
          "title": "Type",
          "description": "The type of the state provider to use. For DataHub use `datahub`",
          "type": "string"
        },
        "config": {
          "title": "Config",
          "description": "The configuration required for initializing the state provider. Default: The datahub_api config if set at pipeline level. Otherwise, the default DatahubClientConfig. See the defaults (https://github.com/datahub-project/datahub/blob/master/metadata-ingestion/src/datahub/ingestion/graph/client.py#L19)."
        }
      },
      "required": [
        "type"
      ],
      "additionalProperties": false
    },
    "StatefulStaleMetadataRemovalConfig": {
      "title": "StatefulStaleMetadataRemovalConfig",
      "description": "Base specialized config for Stateful Ingestion with stale metadata removal capability.",
      "type": "object",
      "properties": {
        "enabled": {
          "title": "Enabled",
          "description": "The type of the ingestion state provider registered with datahub.",
          "default": false,
          "type": "boolean"
        },
        "remove_stale_metadata": {
          "title": "Remove Stale Metadata",
          "description": "Soft-deletes the entities present in the last successful run but missing in the current run with stateful_ingestion enabled.",
          "default": true,
          "type": "boolean"
        }
      },
      "additionalProperties": false
    },
    "BucketDuration": {
      "title": "BucketDuration",
      "description": "An enumeration.",
      "enum": [
        "DAY",
        "HOUR"
      ],
      "type": "string"
    },
    "OAuthIdentityProvider": {
      "title": "OAuthIdentityProvider",
      "description": "An enumeration.",
      "enum": [
        "microsoft",
        "okta"
      ]
    },
    "OAuthConfiguration": {
      "title": "OAuthConfiguration",
      "type": "object",
      "properties": {
        "provider": {
          "description": "Identity provider for oauth.Supported providers are microsoft and okta.",
          "allOf": [
            {
              "$ref": "#/definitions/OAuthIdentityProvider"
            }
          ]
        },
        "authority_url": {
          "title": "Authority Url",
          "description": "Authority url of your identity provider",
          "type": "string"
        },
        "client_id": {
          "title": "Client Id",
          "description": "client id of your registered application",
          "type": "string"
        },
        "scopes": {
          "title": "Scopes",
          "description": "scopes required to connect to snowflake",
          "type": "array",
          "items": {
            "type": "string"
          }
        },
        "use_certificate": {
          "title": "Use Certificate",
          "description": "Do you want to use certificate and private key to authenticate using oauth",
          "default": false,
          "type": "boolean"
        },
        "client_secret": {
          "title": "Client Secret",
          "description": "client secret of the application if use_certificate = false",
          "type": "string",
          "writeOnly": true,
          "format": "password"
        },
        "encoded_oauth_public_key": {
          "title": "Encoded Oauth Public Key",
          "description": "base64 encoded certificate content if use_certificate = true",
          "type": "string"
        },
        "encoded_oauth_private_key": {
          "title": "Encoded Oauth Private Key",
          "description": "base64 encoded private key content if use_certificate = true",
          "type": "string"
        }
      },
      "required": [
        "provider",
        "authority_url",
        "client_id",
        "scopes"
      ],
      "additionalProperties": false
    },
    "DestinationConfig": {
      "title": "DestinationConfig",
      "type": "object",
      "properties": {
        "bucket_duration": {
          "description": "Size of the time window to aggregate usage stats.",
          "default": "DAY",
          "allOf": [
            {
              "$ref": "#/definitions/BucketDuration"
            }
          ]
        },
        "end_time": {
          "title": "End Time",
          "description": "Latest date of lineage/usage to consider. Default: Current time in UTC",
          "type": "string",
          "format": "date-time"
        },
        "start_time": {
          "title": "Start Time",
          "description": "Earliest date of lineage/usage to consider. Default: Last full day in UTC (or hour, depending on `bucket_duration`). You can also specify relative time with respect to end_time such as '-7 days' Or '-7d'.",
          "type": "string",
          "format": "date-time"
        },
        "options": {
          "title": "Options",
          "description": "Any options specified here will be passed to [SQLAlchemy.create_engine](https://docs.sqlalchemy.org/en/14/core/engines.html#sqlalchemy.create_engine) as kwargs.",
          "type": "object"
        },
        "scheme": {
          "title": "Scheme",
          "default": "snowflake",
          "type": "string"
        },
        "username": {
          "title": "Username",
          "description": "Snowflake username.",
          "type": "string"
        },
        "password": {
          "title": "Password",
          "description": "Snowflake password.",
          "type": "string",
          "writeOnly": true,
          "format": "password"
        },
        "private_key": {
          "title": "Private Key",
          "description": "Private key in a form of '-----BEGIN PRIVATE KEY-----\\nprivate-key\\n-----END PRIVATE KEY-----\\n' if using key pair authentication. Encrypted version of private key will be in a form of '-----BEGIN ENCRYPTED PRIVATE KEY-----\\nencrypted-private-key\\n-----END ECNCRYPTED PRIVATE KEY-----\\n' See: https://docs.snowflake.com/en/user-guide/key-pair-auth.html",
          "type": "string"
        },
        "private_key_path": {
          "title": "Private Key Path",
          "description": "The path to the private key if using key pair authentication. Ignored if `private_key` is set. See: https://docs.snowflake.com/en/user-guide/key-pair-auth.html",
          "type": "string"
        },
        "private_key_password": {
          "title": "Private Key Password",
          "description": "Password for your private key. Required if using key pair authentication with encrypted private key.",
          "type": "string",
          "writeOnly": true,
          "format": "password"
        },
        "oauth_config": {
          "title": "Oauth Config",
          "description": "oauth configuration - https://docs.snowflake.com/en/user-guide/python-connector-example.html#connecting-with-oauth",
          "allOf": [
            {
              "$ref": "#/definitions/OAuthConfiguration"
            }
          ]
        },
        "authentication_type": {
          "title": "Authentication Type",
          "description": "The type of authenticator to use when connecting to Snowflake. Supports \"DEFAULT_AUTHENTICATOR\", \"OAUTH_AUTHENTICATOR\", \"EXTERNAL_BROWSER_AUTHENTICATOR\" and \"KEY_PAIR_AUTHENTICATOR\".",
          "default": "DEFAULT_AUTHENTICATOR",
          "type": "string"
        },
        "account_id": {
          "title": "Account Id",
          "description": "Snowflake account identifier. e.g. xy12345,  xy12345.us-east-2.aws, xy12345.us-central1.gcp, xy12345.central-us.azure, xy12345.us-west-2.privatelink. Refer [Account Identifiers](https://docs.snowflake.com/en/user-guide/admin-account-identifier.html#format-2-legacy-account-locator-in-a-region) for more details.",
          "type": "string"
        },
        "warehouse": {
          "title": "Warehouse",
          "description": "Snowflake warehouse.",
          "type": "string"
        },
        "role": {
          "title": "Role",
          "description": "Snowflake role.",
          "type": "string"
        },
        "connect_args": {
          "title": "Connect Args",
          "description": "Connect args to pass to Snowflake SqlAlchemy driver",
          "type": "object"
        },
        "database": {
          "title": "Database",
          "description": "The fivetran connector log database.",
          "type": "string"
        },
        "log_schema": {
          "title": "Log Schema",
          "description": "The fivetran connector log schema.",
          "type": "string"
        }
      },
      "required": [
        "account_id",
        "database",
        "log_schema"
      ],
      "additionalProperties": false
    },
    "FivetranLogConfig": {
      "title": "FivetranLogConfig",
      "type": "object",
      "properties": {
        "destination_platform": {
          "title": "Destination Platform",
          "description": "The destination platform where fivetran connector log tables are dumped.",
          "default": "snowflake",
          "type": "string"
        },
        "destination_config": {
          "title": "Destination Config",
          "description": "If destination platform is 'snowflake', provide snowflake configuration.",
          "allOf": [
            {
              "$ref": "#/definitions/DestinationConfig"
            }
          ]
        }
      },
      "additionalProperties": false
    },
    "AllowDenyPattern": {
      "title": "AllowDenyPattern",
      "description": "A class to store allow deny regexes",
      "type": "object",
      "properties": {
        "allow": {
          "title": "Allow",
          "description": "List of regex patterns to include in ingestion",
          "default": [
            ".*"
          ],
          "type": "array",
          "items": {
            "type": "string"
          }
        },
        "deny": {
          "title": "Deny",
          "description": "List of regex patterns to exclude from ingestion.",
          "default": [],
          "type": "array",
          "items": {
            "type": "string"
          }
        },
        "ignoreCase": {
          "title": "Ignorecase",
          "description": "Whether to ignore case sensitivity during pattern matching.",
          "default": true,
          "type": "boolean"
        }
      },
      "additionalProperties": false
    },
    "PlatformDetail": {
      "title": "PlatformDetail",
      "type": "object",
      "properties": {
        "platform_instance": {
          "title": "Platform Instance",
          "description": "The instance of the platform that all assets produced by this recipe belong to",
          "type": "string"
        },
        "env": {
          "title": "Env",
          "description": "The environment that all assets produced by DataHub platform ingestion source belong to",
          "default": "PROD",
          "type": "string"
        }
      },
      "additionalProperties": false
    }
  }
}

Code Coordinates

• Class Name: datahub.ingestion.source.fivetran.fivetran.FivetranSource
• Browse on GitHub

Questions

If you've got any questions on configuring ingestion for Fivetran, feel free to ping us on our Slack.