Skip to main content

Kafka Connect

Integration Details

This plugin extracts the following:

  • Source and Sink Connectors in Kafka Connect as Data Pipelines
  • For Source connectors - Data Jobs to represent lineage information between source dataset to Kafka topic per {connector_name}:{source_dataset} combination
  • For Sink connectors - Data Jobs to represent lineage information between Kafka topic to destination dataset per {connector_name}:{topic} combination

Concept Mapping

This ingestion source maps the following Source System Concepts to DataHub Concepts:

Source ConceptDataHub ConceptNotes
"kafka-connect"Data Platform
ConnectorDataFlow
Kafka TopicDataset

Current limitations

Works only for

  • Source connectors: JDBC, Debezium, Mongo and Generic connectors with user-defined lineage graph
  • Sink connectors: BigQuery Certified

Important Capabilities

CapabilityStatusNotes
Platform InstanceEnabled by default

CLI based Ingestion

Install the Plugin

pip install 'acryl-datahub[kafka-connect]'

Starter Recipe

Check out the following recipe to get started with ingestion! See below for full configuration options.

For general pointers on writing and running a recipe, see our main recipe guide.

source:
type: "kafka-connect"
config:
# Coordinates
connect_uri: "http://localhost:8083"

# Credentials
username: admin
password: password

# Optional
platform_instance_map:
bigquery: bigquery_platform_instance_id

sink:
# sink configs

Config Details

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

FieldDescription
cluster_name
string
Cluster to ingest from.
Default: connect-cluster
connect_to_platform_map
map(str,map)
connect_uri
string
URI to connect to.
Default: http://localhost:8083/
convert_lineage_urns_to_lowercase
boolean
Whether to convert the urns of ingested lineage dataset to lowercase
Default: False
password
string
Kafka Connect password.
platform_instance
string
The instance of the platform that all assets produced by this recipe belong to
platform_instance_map
map(str,string)
username
string
Kafka Connect username.
env
string
The environment that all assets produced by this connector belong to
Default: PROD
connector_patterns
AllowDenyPattern
regex patterns for connectors to filter for 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
generic_connectors
array(object)
generic_connectors.connector_name 
string
generic_connectors.source_dataset 
string
generic_connectors.source_platform 
string
provided_configs
array(object)
provided_configs.path_key 
string
provided_configs.provider 
string
provided_configs.value 
string
stateful_ingestion
StatefulStaleMetadataRemovalConfig
Base specialized config for Stateful Ingestion with stale metadata removal capability.
stateful_ingestion.enabled
boolean
The type of the ingestion state provider registered with datahub.
Default: False
stateful_ingestion.ignore_new_state
boolean
If set to True, ignores the current checkpoint state.
Default: False
stateful_ingestion.ignore_old_state
boolean
If set to True, ignores the previous checkpoint state.
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

Advanced Configurations

Kafka Connect supports pluggable configuration providers which can load configuration data from external sources at runtime. These values are not available to DataHub ingestion source through Kafka Connect APIs. If you are using such provided configurations to specify connection url (database, etc) in Kafka Connect connector configuration then you will need also add these in provided_configs section in recipe for DataHub to generate correct lineage.

    # Optional mapping of provider configurations if using
provided_configs:
- provider: env
path_key: MYSQL_CONNECTION_URL
value: jdbc:mysql://test_mysql:3306/librarydb

Code Coordinates

  • Class Name: datahub.ingestion.source.kafka_connect.KafkaConnectSource
  • Browse on GitHub

Questions

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