企业官网建设流程全解析

DataHub Kafka 元数据连接器完全指南：从 Topic、Schema Registry 到 Dataset 的实战接入

【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub

本篇技术指南围绕 DataHub 元数据摄入框架（metadata-ingestion）内置的kafka连接器展开，详细讲解如何将 Apache Kafka 集群中的 Topic、Schema Registry（Confluent Schema Registry）中的 Avro / Protobuf / JSON 模式，以及 Confluent Cloud Stream Catalog 中的标签与业务元数据，批量同步为 DataHub 中的 Dataset、Schema Metadata、Dataset Fields 等标准资产。读完本文，你将掌握最小可用 Recipe 的编写、Confluent Cloud 专用配置、自定义 Schema Registry 接入、OAuth 认证回调、Avro 元数据自动映射、多阶段 Schema 解析与消息级数据画像（Data Profiling）等一整套生产可用的接入方案。

连接器能做什么：从 Kafka 到 DataHub 的元数据管道

kafka模块（位于 metadata-ingestion/src/datahub/ingestion/source/kafka/）用于将 Apache Kafka 的元数据摄入 DataHub，专为生产级摄入工作流设计。它是一个开源实现（支持状态 GA，SupportStatus.GA），主要提取三类内容：

• Topic 元数据：通过 Kafka Admin / Consumer 客户端枚举集群中的 Topic，并额外抓取每个 Topic 的分区数、副本因子、retention.ms、cleanup.policy、max.message.bytes等配置作为 Dataset 的自定义属性（custom properties）；
• Schema 元数据：从 Schema Registry 获取每个 Topic 关联的 key/value 模式，支持 Avro、Protobuf 与 JSON 三种模式类型（源码中以SCHEMA_TYPE_AVRO/SCHEMA_TYPE_PROTOBUF/SCHEMA_TYPE_JSON定义，见 kafka_constants.py），并转换为 DataHub 的 SchemaField；
• Confluent Cloud Stream Catalog 元数据（可选）：当目标集群是 Confluent Cloud 时，可同步 Stream Governance 中为 Topic 整理的标签（tags）与业务元数据（business metadata）。

在源码的能力声明中（kafka.py），该连接器通过@capability装饰器明确标注了支持范围：

概念映射：Kafka 世界与 DataHub 数据模型的对应关系

要理解连接器如何组织元数据，首先需要掌握两者的概念映射。以下是文档给出的标准映射关系（原始表格见 README.md）：

从源码看，Dataset 的生成集中在_emit_dataset方法（kafka.py）：Topic 被映射为Dataset实体，subtype=DatasetSubTypes.TOPIC；若开启ingest_schemas_as_entities，Schema Registry 的 Subject 还会以DatasetSubTypes.SCHEMA子类型单独摄入。同时，Topic 的Partitions、Replication Factor以及min.insync.replicas、retention.bytes等配置（枚举定义见 kafka.py 的KafkaTopicConfigKeys）会被写入 Dataset 的自定义属性，为运维侧排查 Topic 配置问题提供了直接依据。

快速上手：最小可用 Recipe

官方在 kafka_recipe.yml 中给出了一个最小可用的 Recipe 骨架：

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

其中connection.bootstrap是 Kafka broker 地址，connection.schema_registry_url是 Schema Registry 端点。platform_instance用于区分同一平台下的多个 Kafka 集群（详见 docs/platform-instances.md）。

执行摄入的命令为：

datahub ingest -c kafka_recipe.yml

摄入前建议先做连接测试，该能力默认开启（@capability(SourceCapability.TEST_CONNECTION, "Enabled by default")）。从 kafka.py 的KafkaConnectionTest实现可见，它通过consumer.list_topics(timeout=10)验证 broker 连通性，通过SchemaRegistryClient(...).get_subjects()验证 Schema Registry 连通性，并将两项结果以basic_connectivity与schema_metadata能力报告的形式返回：

datahub ingest -c kafka_recipe.yml --dry-run

连接 Confluent Cloud：API Key、ACL 与完整 Recipe

前置条件：为 API Key 配置最小 ACL

使用 Confluent Cloud 时，consumer_config.sasl.username和consumer_config.sasl.password使用集群页面的Data Integration -> API Keys中创建的 API 凭证；schema_registry_config.basic.auth.user.info使用 Schema Registry 的 API 凭证（位于Schema Registry -> API credentials）。

创建集群 API Key 时，必须为 Key 关联如下 ACL，DataHub 才能读取 Confluent Cloud 上 Topic 的元数据：

Topic Name = * Permission = ALLOW Operation = DESCRIBE Pattern Type = LITERAL

完整 Recipe

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

${...}形式的占位符支持从环境变量或 DataHub 的 Secret 机制中取值，避免凭证明文落盘（仓库同时提供datahub.configuration.kafka下的连接配置基类与 Secret 脱敏机制，见 kafka_config.py）。

为 Topic 分配 Domain

如需在摄入时为 Topic 自动归属 Domain，可配置domain映射。domain中的键既可以是完整 URN，也可以是裸 Domain ID（如13ae4d85-d955-49fc-8474-9004c663a810），其值使用allow/deny正则模式匹配 Topic 名：

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: - ".*"

注意：目标 Domain 需要先存在于 DataHub 实例中（可参考 docs/domains.md 创建 Domain），摄入时连接器会通过DomainRegistry解析 URN 并写入 Dataset 的 domain 属性（见 kafka.py）。

非默认 Subject 命名策略：topic_subject_map

如果 Schema Registry 使用了非默认的 Subject 命名策略（例如RecordNameStrategy），默认的<topic>-key/<topic>-value查找会失败。此时必须通过topic_subject_map显式声明 Topic 的 key/value 模式与 Subject 名的对应关系：

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"

在 kafka_config.py 中，topic_subject_map的语义被进一步明确：一旦提供，它将覆盖默认的 Subject 解析（即使使用的是TopicNameStrategy或TopicRecordNameStrategy）。

Confluent Cloud Stream Catalog：同步标签与业务元数据

在 Confluent Cloud 上，Stream Governance 中维护的标签与业务元数据存放在Stream Catalog中，而不是直接挂在 Topic 上。开启confluent_catalog配置块即可将它们同步到对应的 DataHub Topic Dataset 上。

Catalog 由 Schema Registry 端点提供，且复用同一个 API Key，因此一个已能访问 Schema Registry 的 Recipe 只需增加一行enabled: true：

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）映射为 DataHub 标签，业务元数据属性（business metadata attributes）映射为 Topic 的自定义属性。可以通过include_tags: false或include_business_metadata: false只取其中一类。对应的完整配置类为KafkaConfluentCatalogConfig（kafka_config.py），其默认值include_tags=True、include_business_metadata=True。

前提条件与限制

• 仅支持 Confluent Cloud，且环境需要购买Stream Governance Advanced套餐。标签与业务元数据属于 Advanced 功能：在 Essentials 套餐上，Catalog API 即使读取也会返回403；自建 Kafka 不存在 Catalog，配置块会被忽略。
• Schema Registry API Key 的角色需要具备 Catalog 读取权限，实践中为环境的DataSteward角色。仅EnvironmentAdmin不够——它可以管理环境，但不具备 Catalog 读取权限。
• 如果 Key 无法读取 Catalog，摄入不会失败，而是跳过 Catalog 元数据并记录一条警告。
• Catalog 覆盖整个环境：如果该环境包含多个 Kafka 集群，同一 Topic 名可能重复出现。这些重名 Topic 会被跳过（伴随警告），除非设置confluent_catalog.cluster_id为当前摄入的集群 ID（如lkc-xxxxx）。
• Catalog不提供 Topic 之间的血缘。Confluent Stream Lineage UI 中展示的生产者/消费者关系图无法通过 API 获取。需要 Connector 到 Topic 血缘的场景，可改用 kafka-connect 连接器。

从源码实现看，Catalog 元数据的应用逻辑位于_apply_catalog_metadata（kafka.py）：当include_tags开启时，Catalog 标签会追加到 Topic 的标签列表；当include_business_metadata开启时，业务元数据通过non_colliding_business_metadata与 broker 侧 Topic 属性做冲突规避后合并进自定义属性，避免覆盖Partitions、Replication Factor等关键属性。值得注意的是，全局标签是替换型（replacement）aspect，如果 Catalog 只被部分读取（is_complete()为假），被遗漏的 Topic 再次摄入时可能丢失原有 Catalog 标签，此时源码会记录一次警告，提示待 Catalog 可完整读取后重跑恢复。

自定义 Schema Registry：实现 KafkaSchemaRegistryBase

Kafka 连接器默认使用 Confluent 的 Kafka Schema Registry 来解析 Topic 的 key/value 模式，原生支持AVRO与PROTOBUF两种模式类型。如果使用的是自定义 Schema Registry，或者模式类型不是 Avro / Protobuf，可以自行实现KafkaSchemaRegistryBase抽象类，并提供get_schema_metadata(topic, platform_urn)方法——该方法接收 Topic 名，返回包含该 Topic 模式的SchemaMetadata对象：

class KafkaSchemaRegistryBase(ABC): @abstractmethod def get_schema_metadata( self, topic: str, platform_urn: str ) -> Optional[SchemaMetadata]: pass

接口定义见 kafka_schema_registry_base.py，其中还包含get_subjects()、_get_subject_for_topic()、get_schema_registry_client()等抽象方法，以及默认的批量模式拉取实现get_schema_and_fields_batch()与build_schema_metadata_with_key()。官方默认实现参考datahub.ingestion.source.confluent_schema_registry::ConfluentSchemaRegistry。

自定义类通过schema_registry_class配置项指定，连接器会使用import_path动态加载（kafka.py）：

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 认证回调：为 Source 与 Sink 配置 OAuth Bearer

Kafka 连接器为 Source（消费者）与 Sink（生产者）均提供了 OAuth 回调支持：

• Source：config.connection.consumer_config.oauth_cb
• Sink：config.connection.producer_config.oauth_cb

回调以<python-module>:<function-name>格式引用 Python 函数。例如oauth:create_token表示create_token定义在oauth.py中，且oauth.py必须在PYTHONPATH中可被导入。

内置回调（推荐）

DataHub 内置了常见场景的 OAuth 回调：

• AWS MSK IAM：datahub_actions.utils.kafka_msk_iam:oauth_cb
• Azure Event Hubs：datahub_actions.utils.kafka_eventhubs_auth:oauth_cb

使用内置回调需安装acryl-datahub-actions包：

pip install acryl-datahub-actions>=1.3.1.2

自定义回调

自定义回调模块需确保 DataHub 进程可访问，例如通过PYTHONPATH=/path/to/your/module:$PYTHONPATH或pip install my-oauth-package。

Kafka Source 示例

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

Kafka Sink 示例（MSK IAM 认证）

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"

从源码看，配置了 OAuth 回调后，连接器会在创建 Consumer / AdminClient 时显式调用一次consumer.poll(timeout=30)触发回调执行（kafka.py），确保后续元数据请求携带有效令牌。

Avro 元数据自动映射：将 Schema 属性转为 Owner、Tag 与 Term

Avro 规范允许 Schema 携带规范未定义的附加属性（arbitrary metadata），业界常用它承载业务元数据。Kafka 连接器可以将这些属性直接转换为 DataHub 的 Owner、Tag 与 Glossary Term。

注意：元数据映射目前仅支持 Avro 模式，且要求这些 Avro 模式已推送到 Schema Registry。同时需要enable_meta_mapping（默认true）开启映射处理。

简单标签：schema_tags_field

如果 Avro Schema 中嵌入了标签列表（顶层或字段级），可用schema_tags_field指定存放标签的字段名（默认tags）。

示例 Avro Schema：

{ "name": "sampleRecord", "type": "record", "tags": ["tag1", "tag2"], "fields": [ { "name": "field_1", "type": "string", "tags": ["tag3", "tag4"] } ] }

config: schema_tags_field: tags

对应源码在 kafka.py：连接器读取 Avro 顶层other_props中schema_tags_field指定的列表，为每个标签加上tag_prefix（默认空字符串）后生成 DataHub 标签。

meta_mapping 与 field_meta_mapping

也可以将特定 Avro 字段精确映射为 Owner、Term 与 Tag：

示例 Avro Schema：

{ "name": "sampleRecord", "type": "record", "owning_team": "@Data-Science", "data_tier": "Bronze", "fields": [ { "name": "field_1", "type": "string", "gdpr": { "pii": true } } ] }

对应的映射配置：

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"

其中meta_mapping针对顶层 Schema 属性，field_meta_mapping针对字段级属性（支持嵌套路径如gdpr.pii），{{ $match }}可引用正则捕获的匹配值。底层实现通过OperationProcessor（kafka.py）执行add_owner/add_term/add_tag三类操作，Owner 来源类型标记为SERVICE；相关指令语义与 dbt 元数据自动映射 的实现一致，那里提供了更丰富的示例可供参考。此外还有strip_user_ids_from_email（从邮箱中剥离用户 ID）与tag_prefix两个辅助配置项。

多阶段 Schema 解析：为缺失 Schema 的 Topic 自动兜底

对于未在 Schema Registry 注册、或使用了非默认命名策略的 Topic，DataHub 提供了多阶段 Schema 解析（schema resolution）。该能力独立于数据画像，二者可分别开关，但共享同一个profiling.max_workers并发配置。

当schema_resolution.enabled为true时（默认关闭），连接器按下述顺序尝试解析：

1. TopicNameStrategy：直接按<topic>-key/<topic>-value查找（最常见）；
2. TopicSubjectMap：使用用户通过topic_subject_map配置的 Topic 到 Subject 映射；
3. RecordNameStrategy：从消息数据中提取记录名，查找<record_name>-key/<record_name>-value；
4. TopicRecordNameStrategy：组合 Topic 与记录名，查找<topic>-<record_name>-key/<record_name>-value；
5. Schema Inference：最后兜底，从消息数据分析推断 Schema。

这一设计保证了与 Confluent Schema Registry 各种命名策略的最大兼容性。上述解析方法在 schema_resolution.py 中均有对应实现，每种策略的结果还附带ResolutionMethod诊断标签（如topic_name_strategy、record_name_strategy、schema_inference，见 kafka_constants.py），摄入日志中会记录每个 Topic 实际命中的解析路径。

配置示例：

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

Schema 推断的采样策略

用于 Schema 推断的消息采样，通过offset_reset_strategy控制读取起点：

• hybrid（默认）：优先尝试latest以追求速度，若未发现近期消息则回退到earliest；
• latest：只读近期消息。最快，但在低频（quiet）Topic 上可能失败；
• earliest：从 Topic 历史起点扫描。最全面，但在大 Topic 上较慢。

性能要点

• 推断出的 Schema 会缓存60 分钟，避免反复采样；
• Worker 数量会根据 CPU 核数与 Topic 数量自动伸缩（默认5 × CPU 核数，见 schema_resolution.py）；
• 将schema_resolution.enabled设为false，则缺失 Schema 时仅产生警告而不自动解析。

配置类定义见 kafka_config.py：sample_timeout_seconds默认2.0秒（单 Topic 采样的时间上限）、max_messages_per_topic默认10条。

数据画像：从消息内容生成字段级统计与样本值

Kafka 连接器支持对消息内容做数据画像（Data Profiling），产出字段级统计与样本值。画像与 Schema 解析相互独立，可只开其一；但两者的并发度都取自profiling.max_workers。

完整配置

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_strategy）

• latest（默认）：从每个分区末尾采样最近的消息；
• random：在分区范围内随机 offset 采样；
• stratified：在 Topic 时间线上均匀分布采样；
• full：处理整个 Topic（仍受sample_size上限约束）。

从源码看，四种策略共享同一套消息解码管线（kafka.py）：latest/random/full仅在选择起始 offset 的函数（_latest_offset/_random_offset/_full_offset）上不同，stratified则通过步长（stride）在分区内等距 seek 采样。样本按batch_size（默认 100）批量消费，并受max_sample_time_seconds（默认 60 秒）时间窗约束。

字段统计与解码细节

画像结果基于每个字段生成DatasetFieldProfile：包括空值计数/比例、唯一值计数/比例、最小值/最大值/均值/中位数/标准差，以及（按需）分位数、去重值频率与直方图；include_field_sample_values会输出实际样本值。字段类型识别依据 Avro 类型归类为NUMERIC/BOOLEAN/DATETIME/STRING/UNKNOWN（分类映射见 kafka_constants.py）。

消息解码时，连接器会识别 Confluent 线格式（magic byte0x00+ 4 字节大端 Schema ID 的 5 字节头，见 kafka_constants.py），用缓存的 Schema 反序列化 Avro 消息，再以nested_field_max_depth限制深度做嵌套字段展平。解码失败的消息会被跳过并计入profiling_samples_skipped/profiling_avro_decode_failures统计，而不会向画像注入伪造字段。

nested_field_max_depth（默认 10）用于防止深度嵌套或循环 JSON 结构引发递归错误；对嵌套复杂的 Topic 建议调低。若 Topic 完全没有 Schema 信息（既不在 Schema Registry，Schema 推断也未开启），画像会自动跳过该 Topic。

画像任务通过ThreadPoolExecutor并行执行（max_workers上限，见 kafka.py），每个 Worker 持有独立的 Consumer 连接，互不阻塞。另外，profiling.operation_config支持按周/月计划（如每周一、每月 15 日）执行低频画像，可参考 SQL 类连接器的操作配置语义（kafka_config.py）。

已知限制

PROTOBUF 模式类型的限制

当前 PROTOBUF 支持存在以下限制：

• 不支持递归类型。
• Protobuf 编译使用进程级全局 descriptor pool：第一个编译某个全限定消息类型的 Topic 会成功；后续嵌入同一类型名的 Topic 会命中duplicate symbol错误，导致该 Topic 没有 Schema 字段（DataHub 记录警告后继续处理）。在大量 Topic 共享公共 proto 类型的环境中，第一个之后的大多数 protobuf Topic 都可能没有 Schema 字段。此时应开启schema_resolution，让这些 Topic 回退到基于消息数据的 Schema 推断。

此外，map 会被表示为消息数组。例如：

message MessageWithMap { map<int, string> map_1 = 1; }

会变成：

message Map1Entry { int key = 1; string value = 2/ } message MessageWithMap { repeated Map1Entry map_1 = 1; }

其他注意点

• Topic 之间的血缘（lineage）不支持；如需 Kafka Connect 场景的血缘，请使用 kafka-connect 连接器。
• 状态化摄入（Stateful Ingestion）仅在为 Source 配置了 Platform Instance 时可用（官方能力注释见 kafka_post.md），用于陈旧元数据的删除检测。

故障排查与性能优化

摄入失败时，首先依次核对：凭证、权限、连通性、范围过滤（scope filters）；然后查看摄入日志中的 Source 专属错误，并据此调整配置。

Schema 解析错误

DataHub 会自动优雅处理 Schema 解析错误并继续处理，无需中断。

Avro 二进制编码错误：avro.errors.InvalidAvroBinaryEncoding: Read 0 bytes, expected 1 bytes

开启 Schema 解析以自动推断：

source: type: kafka config: schema_resolution: enabled: true offset_reset_strategy: "hybrid"

Protobuf 重复符号错误：Couldn't build proto file into descriptor pool: duplicate symbol

DataHub 记录警告并继续处理。存在 Schema 冲突的 Topic 在开启 Schema 解析后会自动使用推断 Schema。

Schema Registry 连接问题：当 Schema Registry 不可达时，可将 Schema 解析作为兜底：

source: type: kafka config: connection: schema_registry_url: "http://localhost:8081" schema_resolution: enabled: true

大型集群性能优化

对拥有大量 Topic 的 Kafka 集群，建议用topic_patterns提前裁剪范围，并合理设置画像与解析并发：

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

topic_patterns的默认值在 kafka_config.py 中定义为allow=[".*"]、deny=["^_.*"]，即默认排除下划线开头的内部 Topic。

大 Topic 内存问题

对消息体较大或吞吐量高的 Topic，降低采样规模与递归深度：

profiling: enabled: true sample_size: 50 nested_field_max_depth: 2

源码侧还为内存保护预设了多项硬性上限（kafka_constants.py）：单个样本值字符串最长 1000 字符、嵌套字典最多展平 100 个键、列表最多 50 个元素、直方图默认 10 个桶、去重值频率最多 10 项，避免单个消息撑爆画像结果或内存。

验证与深入阅读

仓库为 Kafka 连接器配备了完整的单元与集成测试，可作为理解行为的补充资料：

• 单元测试：metadata-ingestion/tests/unit/kafka/test_kafka_source.py、test_kafka_config.py、test_kafka_profiler.py、test_kafka_sampling_strategies.py、test_kafka_schema_inference.py、test_kafka_confluent_catalog.py、test_kafka_protobuf_schema_handling.py
• 集成测试与示例 Recipe：metadata-ingestion/tests/integration/kafka/test_kafka.py（配套kafka_to_file.yml、kafka_catalog_to_file.yml、kafka_to_file_oauth.yml、kafka_without_schemas_to_file.yml等示例）
• 核心实现：kafka.py、kafka_config.py、schema_resolution.py、kafka_profiler.py
• 相关模型：Dataset 实体、Platform Instance 说明

【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub