# Kafka ## Overview [Apache Kafka](https://kafka.apache.org/) is a distributed event streaming platform used for high-throughput, fault-tolerant messaging and real-time data pipelines. Confluent Cloud provides a fully managed Kafka service with an integrated Schema Registry. The DataHub Kafka connector extracts topic metadata and schemas from Kafka clusters and Confluent Schema Registry. It supports Avro, Protobuf, and JSON schemas, multi-stage schema resolution with automatic fallback and inference for topics not registered in the schema registry, and optional data profiling to generate field-level statistics and sample values from message content. The integration also captures stateful deletion detection. ## Concept Mapping | Kafka Concept | DataHub Concept | Notes | | ---------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------- | | Topic | [Dataset](../../metamodel/entities/dataset.md) | Subtype `Topic` | | Schema (Subject) | [Schema Metadata](../../metamodel/entities/dataset.md#schema) | Avro, Protobuf, JSON schemas | | Message Fields | [Dataset Fields](../../metamodel/entities/dataset.md#fields) | Extracted from schemas or inferred | | Kafka Cluster | [Data Platform Instance](../../../platform-instances.md) | When `platform_instance` is configured | | Schema metadata | Tags, [Glossary Terms](../../metamodel/entities/glossaryTerm.md), Owners ([CorpUser](../../metamodel/entities/corpuser.md) / [CorpGroup](../../metamodel/entities/corpGroup.md)) | Optional, Avro only: derived from schema properties when `enable_meta_mapping` is set with `meta_mapping` / `field_meta_mapping` directives | ## Module `kafka` ![GA](https://img.shields.io/badge/support%20status-GA-brightgreen) ### Important Capabilities | Capability | Status | Notes | | ---------- | ------ | ----- | | Column-level Lineage | ❌ | Not supported. | | [Data Profiling](../../../../metadata-ingestion/docs/dev_guides/sql_profiles.md) | ✅ | Optionally enabled via configuration `profiling.enabled`. | | Descriptions | ✅ | Set dataset description to top level doc field for Avro schema. | | [Detect Deleted Entities](../../../../metadata-ingestion/docs/dev_guides/stateful.md#stale-entity-removal) | ✅ | Enabled by default via stateful ingestion. | | Extract Tags | ✅ | Requires Confluent Cloud Stream Governance; enable via `confluent_catalog`. | | [Platform Instance](../../../platform-instances.md) | ✅ | For multiple Kafka clusters, use the platform_instance configuration. | | Schema Metadata | ✅ | Schemas associated with each topic are extracted from the schema registry. Avro and Protobuf (certified), JSON (incubating). Schema references are supported. | | Table-Level Lineage | ❌ | Not supported. If you use Kafka Connect, the kafka-connect source can generate lineage. | | Test Connection | ✅ | Enabled by default. | ### Overview The `kafka` module ingests metadata from Kafka into DataHub. It is intended for production ingestion workflows and module-specific capabilities are documented below. Extract Topics & Schemas from Apache Kafka or Confluent Cloud. This plugin extracts the following: - Topics from the Kafka broker - Schemas associated with each topic from the schema registry (Avro, Protobuf and JSON schemas are supported) - On Confluent Cloud, optionally the tags and business metadata curated for each topic in the Stream Catalog ### Prerequisites Before running ingestion, ensure network connectivity to the source, valid authentication credentials, and read permissions for metadata APIs required by this module. ### Install the Plugin ```shell pip install 'acryl-datahub[kafka]' ``` ### Starter Recipe Check out the following recipe to get started with ingestion! See [below](#config-details) for full configuration options. For general pointers on writing and running a recipe, see our [main recipe guide](../../../../metadata-ingestion/README.md#recipes). ```yaml source: type: "kafka" config: platform_instance: "YOUR_CLUSTER_ID" connection: bootstrap: "broker:9092" schema_registry_url: http://localhost:8081 # Optional: Enable data profiling profiling: enabled: true sample_size: 1000 max_sample_time_seconds: 60 sampling_strategy: "latest" sink: # sink configs ``` ### Config Details #### Options Note that a `.` is used to denote nested fields in the YAML recipe.
| Field | Description | |:--- |:--- | |
convert_urns_to_lowercase
boolean
| Whether to convert dataset urns to lowercase. This value is part of each dataset's URN identity, so it must stay fixed for the life of a deployment. Changing it after data has been ingested re-keys every dataset (e.g. `MyDb.MyTable` becomes `mydb.mytable`); with stateful ingestion enabled the old-cased URNs are then soft-deleted as stale while the new-cased ones are created, producing duplicate or orphaned entities. Pick one value before the first run and leave it unchanged.
Default: False
| |
disable_topic_record_naming_strategy
boolean
| Disables the utilization of the TopicRecordNameStrategy for Schema Registry subjects. For more information, visit: https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html#handling-differences-between-preregistered-and-client-derived-schemas:~:text=io.confluent.kafka.serializers.subject.TopicRecordNameStrategy
Default: False
| |
enable_meta_mapping
boolean
| When enabled, applies the mappings that are defined through the meta_mapping directives.
Default: True
| |
external_url_base
One of string, null
| Base URL for external platform (e.g. Aiven) where topics can be viewed. The topic name will be appended to this base URL.
Default: None
| |
field_meta_mapping
object
| mapping rules that will be executed against field-level schema properties. Refer to the section below on meta automated mappings.
Default: {}
| |
ignore_warnings_on_schema_type
boolean
| Disables warnings reported for non-AVRO/Protobuf value or key schemas if set.
Default: False
| |
ingest_schemas_as_entities
boolean
| Enables ingesting schemas from schema registry as separate entities, in addition to the topics
Default: False
| |
meta_mapping
object
| mapping rules that will be executed against top-level schema properties. Refer to the section below on meta automated mappings.
Default: {}
| |
platform_instance
One of string, null
| The instance of the platform that all assets produced by this recipe belong to. This should be unique within the platform. See https://docs.datahub.com/docs/platform-instances/ for more details.
Default: None
| |
schema_registry_class
string
| The fully qualified implementation class(custom) that implements the KafkaSchemaRegistryBase interface.
Default: datahub.ingestion.source.confluent_schema_registry...
| |
schema_tags_field
string
| The field name in the schema metadata that contains the tags to be added to the dataset.
Default: tags
| |
strip_user_ids_from_email
boolean
| Whether or not to strip email id while adding owners using meta mappings.
Default: False
| |
tag_prefix
string
| Prefix added to tags during ingestion.
Default:
| |
topic_subject_map
map(str,string)
| | |
env
string
| The environment that all assets produced by this connector belong to
Default: PROD
| |
confluent_catalog
KafkaConfluentCatalogConfig
| | |
confluent_catalog.api_key
One of string(password), null
| Schema Registry API key, used for Basic authentication against the Stream Catalog.
Default: None
| |
confluent_catalog.api_secret
One of string(password), null
| Schema Registry API secret, used for Basic authentication against the Stream Catalog.
Default: None
| |
confluent_catalog.cluster_id
One of string, null
| Kafka cluster id, e.g. `lkc-xxxxx`. Only needed when the environment behind this Schema Registry holds more than one Kafka cluster, since the catalog covers the whole environment and topic names can repeat across clusters.
Default: None
| |
confluent_catalog.enabled
boolean
| Query the Confluent Cloud Stream Catalog GraphQL API. Confluent Cloud only, and requires Stream Governance to be enabled on the environment.
Default: False
| |
confluent_catalog.include_business_metadata
boolean
| Emit Confluent Cloud business metadata attributes on topics as DataHub custom properties.
Default: True
| |
confluent_catalog.include_tags
boolean
| Emit Confluent Cloud tags on topics as DataHub tags.
Default: True
| |
confluent_catalog.page_size
integer
| Number of entities to request per Stream Catalog GraphQL page.
Default: 100
| |
confluent_catalog.schema_registry_url
One of string, null
| Schema Registry endpoint, e.g. `https://psrc-xxxxx.us-east-1.aws.confluent.cloud`. The Stream Catalog GraphQL endpoint is derived as `/catalog/graphql`.
Default: None
| |
confluent_catalog.timeout_seconds
integer
| Timeout in seconds for each Stream Catalog GraphQL request.
Default: 30
| |
connection
KafkaConsumerConnectionConfig
| Configuration class for holding connectivity information for Kafka consumers | |
connection.bootstrap
string
|
Default: localhost:9092
| |
connection.client_timeout_seconds
integer
| The request timeout used when interacting with the Kafka APIs.
Default: 60
| |
connection.consumer_config
object
| Extra consumer config serialized as JSON. These options will be passed into Kafka's DeserializingConsumer. See https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#deserializingconsumer and https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md . | |
connection.schema_registry_config
object
| Extra schema registry config serialized as JSON. These options will be passed into Kafka's SchemaRegistryClient. https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html?#schemaregistryclient | |
connection.schema_registry_url
string
| Schema registry URL. Can be overridden with KAFKA_SCHEMAREGISTRY_URL environment variable, or will use DATAHUB_GMS_BASE_PATH if not set. | |
domain
map(str,AllowDenyPattern)
| A class to store allow deny regexes.

Patterns are matched against the start of the string only, not the entire
string - a pattern does not need to match to the end to be considered a match.
For example, the pattern "prod" matches "prod", "prod_east", and "production".
To require an exact match, anchor your pattern explicitly, e.g. "^prod$". | |
domain.`key`.allow
array
| List of regex patterns to include in ingestion. Patterns match from the start of the string only, not the entire string - anchor with '^...$' for an exact match, e.g. '^prod$'.
Default: ['.*']
| |
domain.`key`.allow.string
string
| | |
domain.`key`.ignoreCase
One of boolean, null
| Whether to ignore case sensitivity during pattern matching.
Default: True
| |
domain.`key`.deny
array
| List of regex patterns to exclude from ingestion. Patterns match from the start of the string only, not the entire string - anchor with '^...$' for an exact match, e.g. '^prod$'.
Default: []
| |
domain.`key`.deny.string
string
| | |
schema_resolution
SchemaResolutionFallback
| | |
schema_resolution.enabled
boolean
| Enable comprehensive schema resolution with multiple fallback strategies for topics where schema registry lookup fails.
Default: False
| |
schema_resolution.max_messages_per_topic
integer
| Maximum number of messages to sample per topic for record name extraction and schema inference. Must be positive.
Default: 10
| |
schema_resolution.offset_reset_strategy
Enum
| One of: "earliest", "latest", "hybrid" | |
schema_resolution.sample_timeout_seconds
number
| Maximum time to spend sampling messages from a single topic (in seconds) for record name extraction and schema inference. Must be positive.
Default: 2.0
| |
topic_patterns
AllowDenyPattern
| A class to store allow deny regexes.

Patterns are matched against the start of the string only, not the entire
string - a pattern does not need to match to the end to be considered a match.
For example, the pattern "prod" matches "prod", "prod_east", and "production".
To require an exact match, anchor your pattern explicitly, e.g. "^prod$". | |
topic_patterns.ignoreCase
One of boolean, null
| Whether to ignore case sensitivity during pattern matching.
Default: True
| |
topic_patterns.allow
array
| List of regex patterns to include in ingestion. Patterns match from the start of the string only, not the entire string - anchor with '^...$' for an exact match, e.g. '^prod$'.
Default: ['.*']
| |
topic_patterns.allow.string
string
| | |
topic_patterns.deny
array
| List of regex patterns to exclude from ingestion. Patterns match from the start of the string only, not the entire string - anchor with '^...$' for an exact match, e.g. '^prod$'.
Default: []
| |
topic_patterns.deny.string
string
| | |
profiling
ProfilerConfig
| | |
profiling.batch_size
integer
| Number of messages to fetch in a single batch (for more efficient reading). Must be positive.
Default: 100
| |
profiling.catch_exceptions
boolean
|
Default: True
| |
profiling.enabled
boolean
| Whether profiling should be done.
Default: False
| |
profiling.field_sample_values_limit
integer
| Upper limit for number of sample values to collect for all columns.
Default: 20
| |
profiling.include_field_distinct_count
boolean
| Whether to profile for the number of distinct values for each column.
Default: True
| |
profiling.include_field_distinct_value_frequencies
boolean
| Whether to profile for distinct value frequencies.
Default: False
| |
profiling.include_field_histogram
boolean
| Whether to profile for the histogram for numeric fields.
Default: False
| |
profiling.include_field_max_value
boolean
| Whether to profile for the max value of numeric columns.
Default: True
| |
profiling.include_field_mean_value
boolean
| Whether to profile for the mean value of numeric columns.
Default: True
| |
profiling.include_field_median_value
boolean
| Whether to profile for the median value of numeric columns.
Default: True
| |
profiling.include_field_min_value
boolean
| Whether to profile for the min value of numeric columns.
Default: True
| |
profiling.include_field_null_count
boolean
| Whether to profile for the number of nulls for each column.
Default: True
| |
profiling.include_field_quantiles
boolean
| Whether to profile for the quantiles of numeric columns.
Default: False
| |
profiling.include_field_sample_values
boolean
| Whether to profile for the sample values for all columns.
Default: True
| |
profiling.include_field_stddev_value
boolean
| Whether to profile for the standard deviation of numeric columns.
Default: True
| |
profiling.limit
One of integer, null
| Max number of documents to profile. By default, profiles all documents.
Default: None
| |
profiling.max_distinct_per_statement
integer
| Only used when `query_combiner_flatten_enabled` is on. Maximum COUNT(DISTINCT) columns allowed in one flattened statement. Each one builds a distinct-value tree in server memory, so merging too many trades a scan problem for a memory problem. The default is a starting point, not a measured optimum — raise it if your server has headroom and unique counts dominate profiling time, lower it if profiling causes memory pressure.
Default: 5
| |
profiling.max_number_of_fields_to_profile
One of integer, null
| A positive integer that specifies the maximum number of columns to profile for any table. `None` implies all columns. The cost of profiling goes up significantly as the number of columns to profile goes up.
Default: None
| |
profiling.max_sample_time_seconds
integer
| Maximum time to spend sampling messages in seconds. Must be positive.
Default: 60
| |
profiling.max_workers
integer
| Number of worker threads to use for profiling. Set to 1 to disable.
Default: 20
| |
profiling.nested_field_max_depth
integer
| Maximum recursion depth when flattening nested JSON structures during profiling. Lower values prevent recursion errors but may truncate deeply nested data. Applies to connectors that process dynamic JSON content (e.g., Kafka, MongoDB, Elasticsearch).
Default: 10
| |
profiling.offset
One of integer, null
| Offset in documents to profile. By default, uses no offset.
Default: None
| |
profiling.partition_datetime
One of string(date-time), null
| If specified, profile only the partition which matches this datetime. If not specified, profile the latest partition. Only Bigquery supports this.
Default: None
| |
profiling.partition_profiling_enabled
boolean
| Whether to profile partitioned tables. Only BigQuery and Aws Athena supports this. If enabled, latest partition data is used for profiling.
Default: True
| |
profiling.profile_external_tables
boolean
| Whether to profile external tables. Only Snowflake and Redshift supports this.
Default: False
| |
profiling.profile_if_updated_since_days
One of number, null
| Profile table only if it has been updated since these many number of days. If set to `null`, no constraint of last modified time for tables to profile. Supported in `Snowflake`, `BigQuery`, and `Dremio`. Note: for Dremio this compares against DataHub's last-profiled timestamp (Dremio exposes no table modification time), so it controls profile frequency rather than reacting to upstream change.
Default: None
| |
profiling.profile_nested_fields
boolean
| Whether to profile complex types like structs, arrays and maps.
Default: False
| |
profiling.profile_table_level_only
boolean
| Whether to perform profiling at table-level only, or include column-level profiling as well.
Default: False
| |
profiling.profile_table_row_count_estimate_only
boolean
| Use an approximate query for row count. This will be much faster but slightly less accurate. Only supported for Postgres and MySQL.
Default: False
| |
profiling.profile_table_row_limit
One of integer, null
| Profile tables only if their row count is less than specified count. If set to `null`, no limit on the row count of tables to profile. Supported only in `Snowflake`, `BigQuery`. Supported for `Oracle` based on gathered stats.
Default: 5000000
| |
profiling.profile_table_size_limit
One of integer, null
| Profile tables only if their size is less than specified GBs. If set to `null`, no limit on the size of tables to profile. Supported in `Snowflake`, `BigQuery`, `Databricks`, `Oracle`, and `Teradata`. `Oracle` uses calculated size from gathered stats. `Teradata` uses DBC space accounting.
Default: 5
| |
profiling.profiling_isolation_level
One of Enum, null
| Isolation level for the profiling connection. Defaults to unset, so the connection keeps the driver default and one transaction spans the whole table profile. Set AUTOCOMMIT if profiling is holding Postgres idle-in-transaction (blocking VACUUM) or pinning an InnoDB read view and growing the undo log on MySQL; each profiling SELECT then runs on its own. Under AUTOCOMMIT or READ_COMMITTED, metrics come from different snapshots, so a profile can be internally inconsistent on a concurrently-written table (e.g. uniqueCount > rowCount). On MySQL InnoDB, REPEATABLE_READ keeps a consistent snapshot; on Postgres, READ COMMITTED already takes a fresh snapshot per statement, so AUTOCOMMIT loses no consistency there.
Default: None
| |
profiling.query_combiner_enabled
boolean
| *This feature is still experimental and can be disabled if it causes issues.* Reduces the total number of queries issued and speeds up profiling by dynamically combining SQL queries where possible.
Default: True
| |
profiling.query_combiner_flatten_enabled
boolean
| Flattens same-shape aggregate queries into one flat SELECT per FROM group to reduce full table scans on row stores (e.g. MySQL). Requires `query_combiner_enabled`; has no effect on its own. Off by default. COUNT(DISTINCT) columns are capped per statement to bound server memory.
Default: False
| |
profiling.report_dropped_profiles
boolean
| Whether to report datasets or dataset columns which were not profiled. Set to `True` for debugging purposes.
Default: False
| |
profiling.sample_size
integer
| Number of messages to sample for profiling. Higher values provide more accurate statistics but take longer to process. Must be positive.
Default: 200
| |
profiling.sampling_strategy
Enum
| One of: "latest", "random", "stratified", "full" | |
profiling.turn_off_expensive_profiling_metrics
boolean
| Whether to turn off expensive profiling or not. This turns off profiling for quantiles, distinct_value_frequencies, histogram & sample_values. This also limits maximum number of fields being profiled to 10.
Default: False
| |
profiling.use_sampling
boolean
| Whether to profile column level stats on sample of table. Only BigQuery and Snowflake support this. If enabled, profiling is done on rows sampled from table. Sampling is not done for smaller tables.
Default: True
| |
profiling.operation_config
OperationConfig
| | |
profiling.operation_config.lower_freq_profile_enabled
boolean
| Whether to do profiling at lower freq or not. This does not do any scheduling just adds additional checks to when not to run profiling.
Default: False
| |
profiling.operation_config.profile_date_of_month
One of integer, null
| Number between 1 to 31 for date of month (both inclusive). If not specified, defaults to Nothing and this field does not take affect.
Default: None
| |
profiling.operation_config.profile_day_of_week
One of integer, null
| Number between 0 to 6 for day of week (both inclusive). 0 is Monday and 6 is Sunday. If not specified, defaults to Nothing and this field does not take affect.
Default: None
| |
profiling.tags_to_ignore_sampling
One of array, null
| Fixed list of tags to ignore sampling. Each entry may be a full tag URN (e.g. `urn:li:tag:my_tag`) or just the tag name (e.g. `my_tag`). If not specified, tables will be sampled based on `use_sampling`.
Default: None
| |
profiling.tags_to_ignore_sampling.string
string
| | |
stateful_ingestion
One of StatefulStaleMetadataRemovalConfig, null
|
Default: None
| |
stateful_ingestion.enabled
boolean
| Whether or not to enable stateful ingest. Default: True if a pipeline_name is set and either a datahub-rest sink or `datahub_api` is specified, otherwise False
Default: False
| |
stateful_ingestion.fail_safe_threshold
number
| Prevents large amount of soft deletes & the state from committing from accidental changes to the source configuration if the relative change percent in entities compared to the previous state is above the 'fail_safe_threshold'.
Default: 75.0
| |
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](https://json-schema.org/) for this configuration is inlined below. ```javascript { "$defs": { "AllowDenyPattern": { "additionalProperties": false, "description": "A class to store allow deny regexes.\n\nPatterns are matched against the start of the string only, not the entire\nstring - a pattern does not need to match to the end to be considered a match.\nFor example, the pattern \"prod\" matches \"prod\", \"prod_east\", and \"production\".\nTo require an exact match, anchor your pattern explicitly, e.g. \"^prod$\".", "properties": { "allow": { "default": [ ".*" ], "description": "List of regex patterns to include in ingestion. Patterns match from the start of the string only, not the entire string - anchor with '^...$' for an exact match, e.g. '^prod$'.", "items": { "type": "string" }, "title": "Allow", "type": "array" }, "deny": { "default": [], "description": "List of regex patterns to exclude from ingestion. Patterns match from the start of the string only, not the entire string - anchor with '^...$' for an exact match, e.g. '^prod$'.", "items": { "type": "string" }, "title": "Deny", "type": "array" }, "ignoreCase": { "anyOf": [ { "type": "boolean" }, { "type": "null" } ], "default": true, "description": "Whether to ignore case sensitivity during pattern matching.", "title": "Ignorecase" } }, "title": "AllowDenyPattern", "type": "object" }, "KafkaConfluentCatalogConfig": { "additionalProperties": false, "properties": { "enabled": { "default": false, "description": "Query the Confluent Cloud Stream Catalog GraphQL API. Confluent Cloud only, and requires Stream Governance to be enabled on the environment.", "title": "Enabled", "type": "boolean" }, "schema_registry_url": { "anyOf": [ { "type": "string" }, { "type": "null" } ], "default": null, "description": "Schema Registry endpoint, e.g. `https://psrc-xxxxx.us-east-1.aws.confluent.cloud`. The Stream Catalog GraphQL endpoint is derived as `/catalog/graphql`.", "title": "Schema Registry Url" }, "api_key": { "anyOf": [ { "format": "password", "type": "string", "writeOnly": true }, { "type": "null" } ], "default": null, "description": "Schema Registry API key, used for Basic authentication against the Stream Catalog.", "title": "Api Key" }, "api_secret": { "anyOf": [ { "format": "password", "type": "string", "writeOnly": true }, { "type": "null" } ], "default": null, "description": "Schema Registry API secret, used for Basic authentication against the Stream Catalog.", "title": "Api Secret" }, "page_size": { "default": 100, "description": "Number of entities to request per Stream Catalog GraphQL page.", "title": "Page Size", "type": "integer" }, "timeout_seconds": { "default": 30, "description": "Timeout in seconds for each Stream Catalog GraphQL request.", "title": "Timeout Seconds", "type": "integer" }, "cluster_id": { "anyOf": [ { "type": "string" }, { "type": "null" } ], "default": null, "description": "Kafka cluster id, e.g. `lkc-xxxxx`. Only needed when the environment behind this Schema Registry holds more than one Kafka cluster, since the catalog covers the whole environment and topic names can repeat across clusters.", "title": "Cluster Id" }, "include_tags": { "default": true, "description": "Emit Confluent Cloud tags on topics as DataHub tags.", "title": "Include Tags", "type": "boolean" }, "include_business_metadata": { "default": true, "description": "Emit Confluent Cloud business metadata attributes on topics as DataHub custom properties.", "title": "Include Business Metadata", "type": "boolean" } }, "title": "KafkaConfluentCatalogConfig", "type": "object" }, "KafkaConsumerConnectionConfig": { "additionalProperties": false, "description": "Configuration class for holding connectivity information for Kafka consumers", "properties": { "bootstrap": { "default": "localhost:9092", "title": "Bootstrap", "type": "string" }, "schema_registry_url": { "description": "Schema registry URL. Can be overridden with KAFKA_SCHEMAREGISTRY_URL environment variable, or will use DATAHUB_GMS_BASE_PATH if not set.", "title": "Schema Registry Url", "type": "string" }, "schema_registry_config": { "additionalProperties": true, "description": "Extra schema registry config serialized as JSON. These options will be passed into Kafka's SchemaRegistryClient. https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html?#schemaregistryclient", "title": "Schema Registry Config", "type": "object" }, "client_timeout_seconds": { "default": 60, "description": "The request timeout used when interacting with the Kafka APIs.", "title": "Client Timeout Seconds", "type": "integer" }, "consumer_config": { "additionalProperties": true, "description": "Extra consumer config serialized as JSON. These options will be passed into Kafka's DeserializingConsumer. See https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#deserializingconsumer and https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md .", "title": "Consumer Config", "type": "object" } }, "title": "KafkaConsumerConnectionConfig", "type": "object" }, "OffsetResetStrategy": { "enum": [ "earliest", "latest", "hybrid" ], "title": "OffsetResetStrategy", "type": "string" }, "OperationConfig": { "additionalProperties": false, "properties": { "lower_freq_profile_enabled": { "default": false, "description": "Whether to do profiling at lower freq or not. This does not do any scheduling just adds additional checks to when not to run profiling.", "title": "Lower Freq Profile Enabled", "type": "boolean" }, "profile_day_of_week": { "anyOf": [ { "type": "integer" }, { "type": "null" } ], "default": null, "description": "Number between 0 to 6 for day of week (both inclusive). 0 is Monday and 6 is Sunday. If not specified, defaults to Nothing and this field does not take affect.", "title": "Profile Day Of Week" }, "profile_date_of_month": { "anyOf": [ { "type": "integer" }, { "type": "null" } ], "default": null, "description": "Number between 1 to 31 for date of month (both inclusive). If not specified, defaults to Nothing and this field does not take affect.", "title": "Profile Date Of Month" } }, "title": "OperationConfig", "type": "object" }, "ProfilerConfig": { "additionalProperties": false, "properties": { "enabled": { "default": false, "description": "Whether profiling should be done.", "title": "Enabled", "type": "boolean" }, "operation_config": { "$ref": "#/$defs/OperationConfig", "description": "Experimental feature. To specify operation configs." }, "limit": { "anyOf": [ { "type": "integer" }, { "type": "null" } ], "default": null, "description": "Max number of documents to profile. By default, profiles all documents.", "title": "Limit" }, "offset": { "anyOf": [ { "type": "integer" }, { "type": "null" } ], "default": null, "description": "Offset in documents to profile. By default, uses no offset.", "title": "Offset" }, "profile_table_level_only": { "default": false, "description": "Whether to perform profiling at table-level only, or include column-level profiling as well.", "title": "Profile Table Level Only", "type": "boolean" }, "include_field_null_count": { "default": true, "description": "Whether to profile for the number of nulls for each column.", "title": "Include Field Null Count", "type": "boolean" }, "include_field_distinct_count": { "default": true, "description": "Whether to profile for the number of distinct values for each column.", "title": "Include Field Distinct Count", "type": "boolean" }, "include_field_min_value": { "default": true, "description": "Whether to profile for the min value of numeric columns.", "title": "Include Field Min Value", "type": "boolean" }, "include_field_max_value": { "default": true, "description": "Whether to profile for the max value of numeric columns.", "title": "Include Field Max Value", "type": "boolean" }, "include_field_mean_value": { "default": true, "description": "Whether to profile for the mean value of numeric columns.", "title": "Include Field Mean Value", "type": "boolean" }, "include_field_median_value": { "default": true, "description": "Whether to profile for the median value of numeric columns.", "title": "Include Field Median Value", "type": "boolean" }, "include_field_stddev_value": { "default": true, "description": "Whether to profile for the standard deviation of numeric columns.", "title": "Include Field Stddev Value", "type": "boolean" }, "include_field_quantiles": { "default": false, "description": "Whether to profile for the quantiles of numeric columns.", "title": "Include Field Quantiles", "type": "boolean" }, "include_field_distinct_value_frequencies": { "default": false, "description": "Whether to profile for distinct value frequencies.", "title": "Include Field Distinct Value Frequencies", "type": "boolean" }, "include_field_histogram": { "default": false, "description": "Whether to profile for the histogram for numeric fields.", "title": "Include Field Histogram", "type": "boolean" }, "include_field_sample_values": { "default": true, "description": "Whether to profile for the sample values for all columns.", "title": "Include Field Sample Values", "type": "boolean" }, "max_workers": { "default": 20, "description": "Number of worker threads to use for profiling. Set to 1 to disable.", "title": "Max Workers", "type": "integer" }, "report_dropped_profiles": { "default": false, "description": "Whether to report datasets or dataset columns which were not profiled. Set to `True` for debugging purposes.", "title": "Report Dropped Profiles", "type": "boolean" }, "turn_off_expensive_profiling_metrics": { "default": false, "description": "Whether to turn off expensive profiling or not. This turns off profiling for quantiles, distinct_value_frequencies, histogram & sample_values. This also limits maximum number of fields being profiled to 10.", "title": "Turn Off Expensive Profiling Metrics", "type": "boolean" }, "field_sample_values_limit": { "default": 20, "description": "Upper limit for number of sample values to collect for all columns.", "title": "Field Sample Values Limit", "type": "integer" }, "max_number_of_fields_to_profile": { "anyOf": [ { "exclusiveMinimum": 0, "type": "integer" }, { "type": "null" } ], "default": null, "description": "A positive integer that specifies the maximum number of columns to profile for any table. `None` implies all columns. The cost of profiling goes up significantly as the number of columns to profile goes up.", "title": "Max Number Of Fields To Profile" }, "profile_if_updated_since_days": { "anyOf": [ { "exclusiveMinimum": 0, "type": "number" }, { "type": "null" } ], "default": null, "description": "Profile table only if it has been updated since these many number of days. If set to `null`, no constraint of last modified time for tables to profile. Supported in `Snowflake`, `BigQuery`, and `Dremio`. Note: for Dremio this compares against DataHub's last-profiled timestamp (Dremio exposes no table modification time), so it controls profile frequency rather than reacting to upstream change.", "schema_extra": { "supported_sources": [ "snowflake", "bigquery", "dremio" ] }, "title": "Profile If Updated Since Days" }, "profile_table_size_limit": { "anyOf": [ { "type": "integer" }, { "type": "null" } ], "default": 5, "description": "Profile tables only if their size is less than specified GBs. If set to `null`, no limit on the size of tables to profile. Supported in `Snowflake`, `BigQuery`, `Databricks`, `Oracle`, and `Teradata`. `Oracle` uses calculated size from gathered stats. `Teradata` uses DBC space accounting.", "schema_extra": { "supported_sources": [ "snowflake", "bigquery", "unity-catalog", "oracle", "teradata" ] }, "title": "Profile Table Size Limit" }, "profile_table_row_limit": { "anyOf": [ { "type": "integer" }, { "type": "null" } ], "default": 5000000, "description": "Profile tables only if their row count is less than specified count. If set to `null`, no limit on the row count of tables to profile. Supported only in `Snowflake`, `BigQuery`. Supported for `Oracle` based on gathered stats.", "schema_extra": { "supported_sources": [ "snowflake", "bigquery", "oracle" ] }, "title": "Profile Table Row Limit" }, "profile_table_row_count_estimate_only": { "default": false, "description": "Use an approximate query for row count. This will be much faster but slightly less accurate. Only supported for Postgres and MySQL. ", "schema_extra": { "supported_sources": [ "postgres", "mysql" ] }, "title": "Profile Table Row Count Estimate Only", "type": "boolean" }, "query_combiner_enabled": { "default": true, "description": "*This feature is still experimental and can be disabled if it causes issues.* Reduces the total number of queries issued and speeds up profiling by dynamically combining SQL queries where possible.", "title": "Query Combiner Enabled", "type": "boolean" }, "query_combiner_flatten_enabled": { "default": false, "description": "Flattens same-shape aggregate queries into one flat SELECT per FROM group to reduce full table scans on row stores (e.g. MySQL). Requires `query_combiner_enabled`; has no effect on its own. Off by default. COUNT(DISTINCT) columns are capped per statement to bound server memory.", "title": "Query Combiner Flatten Enabled", "type": "boolean" }, "max_distinct_per_statement": { "default": 5, "description": "Only used when `query_combiner_flatten_enabled` is on. Maximum COUNT(DISTINCT) columns allowed in one flattened statement. Each one builds a distinct-value tree in server memory, so merging too many trades a scan problem for a memory problem. The default is a starting point, not a measured optimum \u2014 raise it if your server has headroom and unique counts dominate profiling time, lower it if profiling causes memory pressure.", "exclusiveMinimum": 0, "title": "Max Distinct Per Statement", "type": "integer" }, "catch_exceptions": { "default": true, "description": "", "title": "Catch Exceptions", "type": "boolean" }, "profiling_isolation_level": { "anyOf": [ { "$ref": "#/$defs/ProfilingIsolationLevel" }, { "type": "null" } ], "default": null, "description": "Isolation level for the profiling connection. Defaults to unset, so the connection keeps the driver default and one transaction spans the whole table profile. Set AUTOCOMMIT if profiling is holding Postgres idle-in-transaction (blocking VACUUM) or pinning an InnoDB read view and growing the undo log on MySQL; each profiling SELECT then runs on its own. Under AUTOCOMMIT or READ_COMMITTED, metrics come from different snapshots, so a profile can be internally inconsistent on a concurrently-written table (e.g. uniqueCount > rowCount). On MySQL InnoDB, REPEATABLE_READ keeps a consistent snapshot; on Postgres, READ COMMITTED already takes a fresh snapshot per statement, so AUTOCOMMIT loses no consistency there.", "schema_extra": { "supported_sources": [ "mysql", "postgres" ] } }, "partition_profiling_enabled": { "default": true, "description": "Whether to profile partitioned tables. Only BigQuery and Aws Athena supports this. If enabled, latest partition data is used for profiling.", "schema_extra": { "supported_sources": [ "athena", "bigquery" ] }, "title": "Partition Profiling Enabled", "type": "boolean" }, "partition_datetime": { "anyOf": [ { "format": "date-time", "type": "string" }, { "type": "null" } ], "default": null, "description": "If specified, profile only the partition which matches this datetime. If not specified, profile the latest partition. Only Bigquery supports this.", "schema_extra": { "supported_sources": [ "bigquery" ] }, "title": "Partition Datetime" }, "use_sampling": { "default": true, "description": "Whether to profile column level stats on sample of table. Only BigQuery and Snowflake support this. If enabled, profiling is done on rows sampled from table. Sampling is not done for smaller tables. ", "schema_extra": { "supported_sources": [ "bigquery", "snowflake" ] }, "title": "Use Sampling", "type": "boolean" }, "sample_size": { "default": 200, "description": "Number of messages to sample for profiling. Higher values provide more accurate statistics but take longer to process. Must be positive.", "exclusiveMinimum": 0, "title": "Sample Size", "type": "integer" }, "profile_external_tables": { "default": false, "description": "Whether to profile external tables. Only Snowflake and Redshift supports this.", "schema_extra": { "supported_sources": [ "redshift", "snowflake" ] }, "title": "Profile External Tables", "type": "boolean" }, "tags_to_ignore_sampling": { "anyOf": [ { "items": { "type": "string" }, "type": "array" }, { "type": "null" } ], "default": null, "description": "Fixed list of tags to ignore sampling. Each entry may be a full tag URN (e.g. `urn:li:tag:my_tag`) or just the tag name (e.g. `my_tag`). If not specified, tables will be sampled based on `use_sampling`.", "title": "Tags To Ignore Sampling" }, "profile_nested_fields": { "default": false, "description": "Whether to profile complex types like structs, arrays and maps. ", "title": "Profile Nested Fields", "type": "boolean" }, "nested_field_max_depth": { "default": 10, "description": "Maximum recursion depth when flattening nested JSON structures during profiling. Lower values prevent recursion errors but may truncate deeply nested data. Applies to connectors that process dynamic JSON content (e.g., Kafka, MongoDB, Elasticsearch).", "exclusiveMinimum": 0, "title": "Nested Field Max Depth", "type": "integer" }, "max_sample_time_seconds": { "default": 60, "description": "Maximum time to spend sampling messages in seconds. Must be positive.", "exclusiveMinimum": 0, "title": "Max Sample Time Seconds", "type": "integer" }, "sampling_strategy": { "$ref": "#/$defs/SamplingStrategy", "default": "latest", "description": "Strategy for sampling messages: 'latest' (from end of topic), 'random' (random offsets), 'stratified' (evenly distributed), 'full' (entire topic, respects sample_size)" }, "batch_size": { "default": 100, "description": "Number of messages to fetch in a single batch (for more efficient reading). Must be positive.", "exclusiveMinimum": 0, "title": "Batch Size", "type": "integer" } }, "title": "ProfilerConfig", "type": "object" }, "ProfilingIsolationLevel": { "enum": [ "AUTOCOMMIT", "READ COMMITTED", "REPEATABLE READ", "SERIALIZABLE" ], "title": "ProfilingIsolationLevel", "type": "string" }, "SamplingStrategy": { "enum": [ "latest", "random", "stratified", "full" ], "title": "SamplingStrategy", "type": "string" }, "SchemaResolutionFallback": { "additionalProperties": false, "properties": { "enabled": { "default": false, "description": "Enable comprehensive schema resolution with multiple fallback strategies for topics where schema registry lookup fails.", "title": "Enabled", "type": "boolean" }, "sample_timeout_seconds": { "default": 2.0, "description": "Maximum time to spend sampling messages from a single topic (in seconds) for record name extraction and schema inference. Must be positive.", "exclusiveMinimum": 0, "title": "Sample Timeout Seconds", "type": "number" }, "offset_reset_strategy": { "$ref": "#/$defs/OffsetResetStrategy", "default": "hybrid", "description": "Where to start reading when sampling messages for schema inference: 'earliest' (scan from beginning), 'latest' (recent messages only), or 'hybrid' (try latest first, fallback to earliest). Distinct from the profiler's `sampling_strategy`." }, "max_messages_per_topic": { "default": 10, "description": "Maximum number of messages to sample per topic for record name extraction and schema inference. Must be positive.", "exclusiveMinimum": 0, "title": "Max Messages Per Topic", "type": "integer" } }, "title": "SchemaResolutionFallback", "type": "object" }, "StatefulStaleMetadataRemovalConfig": { "additionalProperties": false, "description": "Base specialized config for Stateful Ingestion with stale metadata removal capability.", "properties": { "enabled": { "default": false, "description": "Whether or not to enable stateful ingest. Default: True if a pipeline_name is set and either a datahub-rest sink or `datahub_api` is specified, otherwise False", "title": "Enabled", "type": "boolean" }, "remove_stale_metadata": { "default": true, "description": "Soft-deletes the entities present in the last successful run but missing in the current run with stateful_ingestion enabled.", "title": "Remove Stale Metadata", "type": "boolean" }, "fail_safe_threshold": { "default": 75.0, "description": "Prevents large amount of soft deletes & the state from committing from accidental changes to the source configuration if the relative change percent in entities compared to the previous state is above the 'fail_safe_threshold'.", "maximum": 100.0, "minimum": 0.0, "title": "Fail Safe Threshold", "type": "number" } }, "title": "StatefulStaleMetadataRemovalConfig", "type": "object" } }, "additionalProperties": false, "properties": { "convert_urns_to_lowercase": { "default": false, "description": "Whether to convert dataset urns to lowercase. This value is part of each dataset's URN identity, so it must stay fixed for the life of a deployment. Changing it after data has been ingested re-keys every dataset (e.g. `MyDb.MyTable` becomes `mydb.mytable`); with stateful ingestion enabled the old-cased URNs are then soft-deleted as stale while the new-cased ones are created, producing duplicate or orphaned entities. Pick one value before the first run and leave it unchanged.", "title": "Convert Urns To Lowercase", "type": "boolean" }, "env": { "default": "PROD", "description": "The environment that all assets produced by this connector belong to", "title": "Env", "type": "string" }, "platform_instance": { "anyOf": [ { "type": "string" }, { "type": "null" } ], "default": null, "description": "The instance of the platform that all assets produced by this recipe belong to. This should be unique within the platform. See https://docs.datahub.com/docs/platform-instances/ for more details.", "title": "Platform Instance" }, "stateful_ingestion": { "anyOf": [ { "$ref": "#/$defs/StatefulStaleMetadataRemovalConfig" }, { "type": "null" } ], "default": null }, "connection": { "$ref": "#/$defs/KafkaConsumerConnectionConfig" }, "topic_patterns": { "$ref": "#/$defs/AllowDenyPattern" }, "domain": { "additionalProperties": { "$ref": "#/$defs/AllowDenyPattern" }, "default": {}, "description": "A map of domain names to allow deny patterns. Domains can be urn-based (`urn:li:domain:13ae4d85-d955-49fc-8474-9004c663a810`) or bare (`13ae4d85-d955-49fc-8474-9004c663a810`).", "title": "Domain", "type": "object" }, "topic_subject_map": { "additionalProperties": { "type": "string" }, "default": {}, "description": "Provides the mapping for the `key` and the `value` schemas of a topic to the corresponding schema registry subject name. Each entry of this map has the form `-key`:`` and `-value`:`` for the key and the value schemas associated with the topic, respectively. This parameter is mandatory when the [RecordNameStrategy](https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html#how-the-naming-strategies-work) is used as the subject naming strategy in the kafka schema registry. NOTE: When provided, this overrides the default subject name resolution even when the `TopicNameStrategy` or the `TopicRecordNameStrategy` are used.", "title": "Topic Subject Map", "type": "object" }, "schema_registry_class": { "default": "datahub.ingestion.source.confluent_schema_registry.ConfluentSchemaRegistry", "description": "The fully qualified implementation class(custom) that implements the KafkaSchemaRegistryBase interface.", "title": "Schema Registry Class", "type": "string" }, "schema_tags_field": { "default": "tags", "description": "The field name in the schema metadata that contains the tags to be added to the dataset.", "title": "Schema Tags Field", "type": "string" }, "enable_meta_mapping": { "default": true, "description": "When enabled, applies the mappings that are defined through the meta_mapping directives.", "title": "Enable Meta Mapping", "type": "boolean" }, "meta_mapping": { "additionalProperties": true, "default": {}, "description": "mapping rules that will be executed against top-level schema properties. Refer to the section below on meta automated mappings.", "title": "Meta Mapping", "type": "object" }, "field_meta_mapping": { "additionalProperties": true, "default": {}, "description": "mapping rules that will be executed against field-level schema properties. Refer to the section below on meta automated mappings.", "title": "Field Meta Mapping", "type": "object" }, "strip_user_ids_from_email": { "default": false, "description": "Whether or not to strip email id while adding owners using meta mappings.", "title": "Strip User Ids From Email", "type": "boolean" }, "tag_prefix": { "default": "", "description": "Prefix added to tags during ingestion.", "title": "Tag Prefix", "type": "string" }, "ignore_warnings_on_schema_type": { "default": false, "description": "Disables warnings reported for non-AVRO/Protobuf value or key schemas if set.", "title": "Ignore Warnings On Schema Type", "type": "boolean" }, "schema_resolution": { "$ref": "#/$defs/SchemaResolutionFallback", "description": "Configuration for comprehensive schema resolution with multiple fallback strategies." }, "disable_topic_record_naming_strategy": { "default": false, "description": "Disables the utilization of the TopicRecordNameStrategy for Schema Registry subjects. For more information, visit: https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html#handling-differences-between-preregistered-and-client-derived-schemas:~:text=io.confluent.kafka.serializers.subject.TopicRecordNameStrategy", "title": "Disable Topic Record Naming Strategy", "type": "boolean" }, "ingest_schemas_as_entities": { "default": false, "description": "Enables ingesting schemas from schema registry as separate entities, in addition to the topics", "title": "Ingest Schemas As Entities", "type": "boolean" }, "external_url_base": { "anyOf": [ { "type": "string" }, { "type": "null" } ], "default": null, "description": "Base URL for external platform (e.g. Aiven) where topics can be viewed. The topic name will be appended to this base URL.", "title": "External Url Base" }, "profiling": { "$ref": "#/$defs/ProfilerConfig", "description": "Settings for message sampling and profiling" }, "confluent_catalog": { "$ref": "#/$defs/KafkaConfluentCatalogConfig", "description": "Read topic tags and business metadata from the Confluent Cloud Stream Catalog. Connection details default to the Schema Registry ones already set under `connection`." } }, "title": "KafkaSourceConfig", "type": "object" } ``` ### Capabilities :::note Stateful Ingestion is available only when a Platform Instance is assigned to this source. ::: Use the **Important Capabilities** table above as the source of truth for supported features and whether additional configuration is required. #### Connecting to Confluent Cloud If using Confluent Cloud you can use a recipe like this. In this `consumer_config.sasl.username` and `consumer_config.sasl.password` are the API credentials that you get (in the Confluent UI) from your cluster -> Data Integration -> API Keys. `schema_registry_config.basic.auth.user.info` has API credentials for Confluent schema registry which you get (in Confluent UI) from Schema Registry -> API credentials. When creating API Key for the cluster ensure that the ACLs associated with the key are set like below. This is required for DataHub to read topic metadata from topics in Confluent Cloud. ``` Topic Name = * Permission = ALLOW Operation = DESCRIBE Pattern Type = LITERAL ``` ```yml source: type: "kafka" config: platform_instance: "YOUR_CLUSTER_ID" connection: bootstrap: "abc-defg.eu-west-1.aws.confluent.cloud:9092" consumer_config: security.protocol: "SASL_SSL" sasl.mechanism: "PLAIN" sasl.username: "${CLUSTER_API_KEY_ID}" sasl.password: "${CLUSTER_API_KEY_SECRET}" schema_registry_url: "https://abc-defgh.us-east-2.aws.confluent.cloud" schema_registry_config: basic.auth.user.info: "${REGISTRY_API_KEY_ID}:${REGISTRY_API_KEY_SECRET}" sink: # sink configs ``` If you are trying to add domains to your topics you can use a configuration like below. ```yml source: type: "kafka" config: # ...connection block domain: "urn:li:domain:13ae4d85-d955-49fc-8474-9004c663a810": allow: - ".*" "urn:li:domain:d6ec9868-6736-4b1f-8aa6-fee4c5948f17": deny: - ".*" ``` Note that the `domain` in config above can be either an _urn_ or a domain _id_ (i.e. `urn:li:domain:13ae4d85-d955-49fc-8474-9004c663a810` or simply `13ae4d85-d955-49fc-8474-9004c663a810`). The Domain should exist in your DataHub instance before ingesting data into the Domain. To create a Domain on DataHub, check out the [Domains User Guide](/docs/domains/). If you are using a non-default subject naming strategy in the schema registry, such as [RecordNameStrategy](https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html#how-the-naming-strategies-work), the mapping for the topic's key and value schemas to the schema registry subject names should be provided via `topic_subject_map` as shown in the configuration below. ```yml source: type: "kafka" config: # ...connection block # Defines the mapping for the key & value schemas associated with a topic & the subject name registered with the # kafka schema registry. topic_subject_map: # Defines both key & value schema for topic 'my_topic_1' "my_topic_1-key": "io.acryl.Schema1" "my_topic_1-value": "io.acryl.Schema2" # Defines only the value schema for topic 'my_topic_2' (the topic doesn't have a key schema). "my_topic_2-value": "io.acryl.Schema3" ``` #### Confluent Cloud Stream Catalog On Confluent Cloud, tags and business metadata curated in Stream Governance are held in the Stream Catalog rather than on the topics themselves. Enable the `confluent_catalog` block to bring them across onto the corresponding DataHub topic datasets. The catalog is served from the Schema Registry endpoint and accepts the same API key, so a recipe that already reaches Schema Registry needs nothing beyond `enabled: true`: ```yml source: type: "kafka" config: connection: bootstrap: "abc-defg.eu-west-1.aws.confluent.cloud:9092" schema_registry_url: "https://abc-defgh.us-east-2.aws.confluent.cloud" schema_registry_config: basic.auth.user.info: "${REGISTRY_API_KEY_ID}:${REGISTRY_API_KEY_SECRET}" confluent_catalog: enabled: true ``` Confluent tags become DataHub tags and business metadata attributes become custom properties on the topic. Set `include_tags` or `include_business_metadata` to `false` to take only one of the two. Requirements and limitations: - **Confluent Cloud only**, and the environment needs the **Stream Governance Advanced** package. Tags and business metadata are an Advanced feature: on Essentials the catalog API returns `403` even for reads. The catalog does not exist on self-managed Kafka, and the block is ignored elsewhere. - The Schema Registry API key needs a role that grants catalog read access — in practice **DataSteward** on the environment. `EnvironmentAdmin` alone is not enough: it administers the environment but does not carry the catalog read permission. - If the key cannot read the catalog, ingestion continues without catalog metadata and records a warning rather than failing. - The catalog covers a whole **environment**, so if that environment holds more than one Kafka cluster the same topic name can appear twice. Those topics are skipped, with a warning, unless you set `confluent_catalog.cluster_id` to the cluster this recipe ingests (e.g. `lkc-xxxxx`). - The catalog carries **no lineage between topics**. The producer/consumer graph shown in Confluent's Stream Lineage UI is not exposed by the API. Connector-to-topic lineage is available through the [`kafka-connect`](kafka-connect.md) source for environments that use Kafka Connect. #### Custom Schema Registry The Kafka Source uses the schema registry to figure out the schema associated with both `key` and `value` for the topic. By default it uses the [Confluent's Kafka Schema registry](https://docs.confluent.io/platform/current/schema-registry/index.html) and supports the `AVRO` and `PROTOBUF` schema types. If you're using a custom schema registry, or you are using schema type other than `AVRO` or `PROTOBUF`, then you can provide your own custom implementation of the `KafkaSchemaRegistryBase` class, and implement the `get_schema_metadata(topic, platform_urn)` method that given a topic name would return object of `SchemaMetadata` containing schema for that topic. Please refer `datahub.ingestion.source.confluent_schema_registry::ConfluentSchemaRegistry` for sample implementation of this class. ```python class KafkaSchemaRegistryBase(ABC): @abstractmethod def get_schema_metadata( self, topic: str, platform_urn: str ) -> Optional[SchemaMetadata]: pass ``` The custom schema registry class can be configured using the `schema_registry_class` config param of the `kafka` source as shown below. ```yaml source: type: "kafka" config: # Set the custom schema registry implementation class schema_registry_class: "datahub.ingestion.source.confluent_schema_registry.ConfluentSchemaRegistry" # Coordinates connection: bootstrap: "broker:9092" schema_registry_url: http://localhost:8081 ``` #### OAuth Callback The OAuth callback function can be set up for both Kafka sources (consumers) and sinks (producers): - For sources: `config.connection.consumer_config.oauth_cb` - For sinks: `config.connection.producer_config.oauth_cb` You need to specify a Python function reference in the format <python-module>:<function-name>. For example, in the configuration `oauth:create_token`, `create_token` is a function defined in `oauth.py`, and `oauth.py` must be accessible in the PYTHONPATH. ##### Deploying Custom OAuth Callbacks **For Built-in Callbacks (Recommended):** DataHub includes pre-built OAuth callbacks for common use cases: - **AWS MSK IAM**: `datahub_actions.utils.kafka_msk_iam:oauth_cb` - **Azure Event Hubs**: `datahub_actions.utils.kafka_eventhubs_auth:oauth_cb` **Important:** To use these built-in callbacks, you must install the `acryl-datahub-actions` package: ```bash pip install acryl-datahub-actions>=1.3.1.2 ``` **For Custom OAuth Callbacks:** If you need to implement a custom OAuth callback, you must ensure your Python module is accessible to the DataHub process, e.g. adding it via `PYTHONPATH=/path/to/your/module:$PYTHONPATH` or `pip install my-oauth-package`. **Example for Kafka Source:** ```yaml source: type: "kafka" config: # Set the custom schema registry implementation class schema_registry_class: "datahub.ingestion.source.confluent_schema_registry.ConfluentSchemaRegistry" # Coordinates connection: bootstrap: "broker:9092" schema_registry_url: http://localhost:8081 consumer_config: security.protocol: "SASL_PLAINTEXT" sasl.mechanism: "OAUTHBEARER" oauth_cb: "oauth:create_token" # sink configs ``` **Example for Kafka Sink (e.g., MSK IAM authentication):** ```yaml sink: type: "datahub-kafka" config: connection: bootstrap: "b-1.msk.us-west-2.amazonaws.com:9098" schema_registry_url: "http://datahub-gms:8080/schema-registry/api/" producer_config: security.protocol: "SASL_SSL" sasl.mechanism: "OAUTHBEARER" sasl.oauthbearer.method: "default" oauth_cb: "datahub_actions.utils.kafka_msk_iam:oauth_cb" ``` #### Enriching DataHub metadata with automated meta mapping :::note Meta mapping is currently only available for Avro schemas, and requires that those Avro schemas are pushed to the schema registry. ::: Avro schemas are permitted to have additional attributes not defined by the specification as arbitrary metadata. A common pattern is to utilize this for business metadata. The Kafka source has the ability to transform this directly into DataHub Owners, Tags and Terms. ##### Simple tags If you simply have a list of tags embedded into an Avro schema (either at the top-level or for an individual field), you can use the `schema_tags_field` config. Example Avro schema: ```json { "name": "sampleRecord", "type": "record", "tags": ["tag1", "tag2"], "fields": [ { "name": "field_1", "type": "string", "tags": ["tag3", "tag4"] } ] } ``` The name of the field containing a list of tags can be configured with the `schema_tags_field` property: ```yaml config: schema_tags_field: tags ``` ##### meta mapping You can also map specific Avro fields into Owners, Tags and Terms using meta mapping. Example Avro schema: ```json { "name": "sampleRecord", "type": "record", "owning_team": "@Data-Science", "data_tier": "Bronze", "fields": [ { "name": "field_1", "type": "string", "gdpr": { "pii": true } } ] } ``` This can be mapped to DataHub metadata with `meta_mapping` config: ```yaml config: meta_mapping: owning_team: match: "^@(.*)" operation: "add_owner" config: owner_type: group data_tier: match: "Bronze|Silver|Gold" operation: "add_term" config: term: "{{ $match }}" field_meta_mapping: gdpr.pii: match: true operation: "add_tag" config: tag: "pii" ``` The underlying implementation is similar to [dbt meta mapping](/docs/generated/ingestion/sources/dbt#dbt-meta-automated-mappings), which has more detailed examples that can be used for reference. #### Schema Resolution DataHub provides multi-stage schema resolution for topics whose schemas are not registered in the schema registry, or that use non-default naming strategies. This feature is **independent** of data profiling. When `schema_resolution.enabled` is true, DataHub attempts to resolve schemas using the following stages in order: 1. **TopicNameStrategy**: Direct lookup using `-key/value` pattern (most common) 2. **TopicSubjectMap**: User-defined topic-to-subject mappings via `topic_subject_map` config 3. **RecordNameStrategy**: Extract record names from message data and lookup `-key/value` 4. **TopicRecordNameStrategy**: Combine topic + record name: `--key/value` 5. **Schema Inference**: As final fallback, infer schema from message data analysis This ensures maximum compatibility with different Confluent Schema Registry [naming strategies](https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html#how-the-naming-strategies-work). ```yaml source: type: kafka config: schema_resolution: enabled: true # disabled by default sample_timeout_seconds: 2.0 offset_reset_strategy: "hybrid" # "earliest", "latest", or "hybrid" max_messages_per_topic: 10 profiling: max_workers: 20 # controls parallelization for both profiling and schema resolution nested_field_max_depth: 5 ``` **Sampling strategies for schema inference:** - **`hybrid` (default)**: Tries `latest` first for speed, falls back to `earliest` if no recent messages found. - **`latest`**: Only reads recent messages. Fastest but may fail on quiet topics. - **`earliest`**: Scans from the beginning of topic history. Most comprehensive but slower on large topics. **Performance notes:** - Inferred schemas are cached for 60 minutes to avoid re-sampling. - Worker count scales automatically based on CPU cores and topic count. - Set `schema_resolution.enabled: false` to receive warnings for missing schemas instead of automatic resolution. #### Data Profiling The Kafka source supports data profiling of message content to generate field-level statistics and sample values. Profiling and schema resolution are enabled independently — either can be turned on without the other. Note, however, that both features draw their parallelism from the same `profiling.max_workers` setting. ```yaml source: type: "kafka" config: profiling: enabled: true sample_size: 200 # messages to sample per topic max_sample_time_seconds: 60 sampling_strategy: "latest" # latest, random, stratified, or full max_workers: 4 batch_size: 100 # Field-level statistics include_field_null_count: true include_field_distinct_count: true include_field_min_value: true include_field_max_value: true include_field_mean_value: true include_field_median_value: true include_field_stddev_value: true include_field_quantiles: false # expensive, disabled by default include_field_distinct_value_frequencies: false # expensive include_field_histogram: false # expensive include_field_sample_values: true # Nested field handling profile_nested_fields: true nested_field_max_depth: 10 # Scheduled profiling (optional) operation_config: lower_freq_profile_enabled: false profile_day_of_week: 1 # Monday=0, Sunday=6 profile_date_of_month: 15 ``` **Sampling strategies:** - **`latest`** (default): Samples the most recent messages from the end of each partition. - **`random`**: Samples messages from random offsets across partitions. - **`stratified`**: Evenly distributes samples across the topic timeline. - **`full`**: Processes the entire topic (respects `sample_size` limit). The `nested_field_max_depth` setting (default: 10) prevents recursion errors on deeply nested or circular JSON structures. Reduce it for topics with complex nested messages. ### Limitations Module behavior is constrained by source APIs, permissions, and metadata exposed by the platform. Refer to capability notes for unsupported or conditional features. #### `PROTOBUF` Schema Type Limitations The current implementation of the support for `PROTOBUF` schema type has the following limitations: - Recursive types are not supported. - Protobuf compilation uses a **process-global descriptor pool**. The first topic that compiles a given fully-qualified message type succeeds; any later topic that embeds the same type name hits a `duplicate symbol` error and is left with no schema fields (DataHub logs a warning and continues). On estates where many topics share common proto types, most protobuf topics after the first will therefore have no schema fields. Enable `schema_resolution` so those topics fall back to schema inference from message data. In addition to this, maps are represented as arrays of messages. The following message, ``` message MessageWithMap { map map_1 = 1; } ``` becomes: ``` message Map1Entry { int key = 1; string value = 2/ } message MessageWithMap { repeated Map1Entry map_1 = 1; } ``` ### Troubleshooting If ingestion fails, validate credentials, permissions, connectivity, and scope filters first. Then review ingestion logs for source-specific errors and adjust configuration accordingly. #### Schema Parsing Errors DataHub automatically handles schema parsing errors gracefully and continues processing. ##### Avro Binary Encoding Errors **Error:** `avro.errors.InvalidAvroBinaryEncoding: Read 0 bytes, expected 1 bytes` Enable schema resolution to automatically infer schemas: ```yaml source: type: kafka config: schema_resolution: enabled: true offset_reset_strategy: "hybrid" ``` ##### Protobuf Duplicate Symbol Errors **Error:** `Couldn't build proto file into descriptor pool: duplicate symbol` DataHub logs warnings and continues processing. Topics with schema conflicts will use inferred schemas if schema resolution is enabled. ##### Schema Registry Connection Issues If schema registry is unavailable, enable schema resolution as a fallback: ```yaml source: type: kafka config: connection: schema_registry_url: "http://localhost:8081" schema_resolution: enabled: true ``` #### Performance Optimization For large Kafka clusters with many topics: ```yaml source: type: kafka config: topic_patterns: allow: ["prod_.*", "analytics_.*"] deny: [".*_temp", ".*_test"] profiling: enabled: true max_workers: 20 # controls both profiling and schema resolution parallelization sample_size: 100 nested_field_max_depth: 10 schema_resolution: enabled: true sample_timeout_seconds: 1.0 ``` #### Memory Issues with Large Topics For topics with large messages or high volume, reduce the sample size and recursion depth: ```yaml profiling: enabled: true sample_size: 50 nested_field_max_depth: 2 ``` ### Code Coordinates - Class Name: `datahub.ingestion.source.kafka.kafka.KafkaSource` - Browse on [GitHub](https://github.com/datahub-project/datahub/blob/master/metadata-ingestion/src/datahub/ingestion/source/kafka/kafka.py) :::tip Questions? If you've got any questions on configuring ingestion for Kafka, feel free to ping us on [our Slack](https://datahub.com/slack). ::: :::note 💡 **Contributing to this documentation** This page is auto-generated from the underlying source code. To make changes, please edit the relevant source files in the [metadata-ingestion](https://github.com/datahub-project/datahub/tree/master/metadata-ingestion) directory. **Tip:** For quick typo fixes or documentation updates, you can click the ✏️ **Edit** icon directly in the GitHub UI to open a Pull Request. For larger changes and PR naming conventions, please refer to our [Contributing Guide](/docs/contributing). :::