DataHub Cassandra 数据源接入指南Keyspace/Table/View 元数据摄取与配置详解【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahubDataHub 官方提供的 Cassandra 集成用于将 Cassandra 集群中的核心元数据数据集/表/视图、Schema 字段、容器摄取进 DataHub并支持基于状态化摄取Stateful Ingestion的删除检测。本文以仓库中的 Cassandra Source 文档 为主线结合 cassandra_pre.md、cassandra_post.md 与完整 示例 Recipe并深入 源码 讲解其原理帮助你完成 Cassandra含 DataStax Astra DB到 DataHub 的元数据同步。Cassandra 集成概览Cassandra 是一个用于存储和查询分析型analytical或操作型operational数据的分布式数据平台。DataHub 的 Cassandra 集成覆盖以下核心元数据实体数据集 / 表 / 视图Dataset/Table/ViewSchema 字段SchemaField容器Container状态化删除检测Stateful Deletion Detection从源码装饰器可以确认该 Source 的能力声明见 cassandra.py能力SourceCapability声明CONTAINERS容器Enabled by defaultSCHEMA_METADATASchema 元数据Enabled by defaultPLATFORM_INSTANCE平台实例Enabled by defaultDELETION_DETECTION删除检测Enabled by default via stateful ingestion该 Source 的 SupportStatus 为BETA用于生产摄取工作流相关模块级能力请以后文及官方文档为准。概念映射Concept Mapping原文档明确指出Cassandra 专属的概念映射细节仍待完善the specific concept mapping is still pending以下是 DataHub 中通用的概念映射关系同样适用于 Cassandra源概念Source ConceptDataHub 概念说明Platform/account/project scopePlatform Instance, Container在平台上下文内组织资产。Core technical asset例如 table/view/topic/fileDataset主要摄取的资产类型。Schema fields / columnsSchemaField支持 Schema 提取时包含。Ownership and collaboration principalsCorpUser, CorpGroup由支持所有权与身份元数据的模块发出。Dependencies and processing relationshipsLineage edges支持并启用了血缘提取时可用。结合源码实现Cassandra 集成实际落地了上述映射的绝大部分具体如下Keyspace → Container在 cassandra.py 中每个 keyspace 会被生成一个Container其subtype为DatasetContainerSubTypes.KEYSPACE并附带durable_writes与replication两个额外属性extra_properties。Table / View → Dataset表被生成为DatasetSubTypes.TABLE类型的数据集物化视图被生成为DatasetSubTypes.VIEW类型的数据集详见 cassandra.py 与 cassandra.py。Column → SchemaField每一列通过CassandraToSchemaFieldConverter转换成SchemaField见 cassandra_utils.py。物化视图 → Lineage edges视图会基于其base_table_name生成与基表之间的血缘边并附带字段级fine-grained血缘这在大多数通用 Source 中并不常见是 Cassandra 集成的亮点之一见下文“视图血缘”章节。摄取前置条件与权限准备在运行摄取之前需要确保到数据源的网络连通性有效的认证凭据该模块所需元数据 API 的读取权限。本模块支持DataStax Astra DB与Cassandra Enterprise EditionEE。获取连接所需信息Astra DB云登录 Astra DB Console进入Organization Settings Token Management生成具有读取所需权限的Application Token从 Astra DB Console 下载Secure Connect Bundle。Cassandra EE本地/自建准备对所需 keyspaces 具有读权限的用户名username与密码password。权限要求用户或 Token 必须具备SELECT权限以支持访问系统 keyspace如system_schema中的元数据获取 keyspaces、tables、columns、views 的信息若启用了数据剖析data profiling还需对数据表执行SELECT操作。源码中实际使用的系统查询均针对system_schema定义在 cassandra_api.py-- 获取所有 keyspace SELECT * FROM system_schema.keyspaces -- 获取某 keyspace 下所有表 SELECT * FROM system_schema.tables WHERE keyspace_name %s -- 获取某表的全部列 SELECT * FROM system_schema.columns WHERE keyspace_name %s AND table_name %s -- 获取某 keyspace 下所有视图 SELECT * FROM system_schema.views WHERE keyspace_name %s -- 行数统计profiling 用 SELECT COUNT(*) AS row_count FROM {keyspace}.{table}验证数据库访问Astra DB确保 Secure Connect Bundle 被正确配置与使用Cassandra 开源版确保contact point与port可访问。:::caution 注意 启用 profiling 时务必对采样的行数设置上限。对超大表不加限制地进行剖析可能导致资源过度消耗与性能下降。 ::::::note 提示 对于 Astra DB 云配置必须在配置中指定 Secure Connect Bundle 路径因此请使用 CLI 方式向 DataHub 摄取元数据。 :::配置详解与完整 RecipeCassandra Source 的配置类定义在 cassandra_config.py下面结合示例 Recipe cassandra_recipe.yml 逐一说明。完整配置示例可直接运行source: type: cassandra config: # Credentials for on prem cassandra contact_point: localhost port: 9042 username: admin password: password # SSL Configuration (optional) #ssl_ca_certs: /path/to/ca-certificate.pem #ssl_certfile: /path/to/client-certificate.pem #ssl_keyfile: /path/to/client-private-key.pem #ssl_version: TLS_CLIENT # Options: TLS_CLIENT, TLSv1, TLSv1_1, TLSv1_2, TLSv1_3 # Or # Credentials Astra Cloud #cloud_config: # secure_connect_bundle: Path to Secure Connect Bundle (.zip) # token: Application Token # Optional Allow / Deny extraction of particular keyspaces. keyspace_pattern: allow: [.*] # Optional Allow / Deny extraction of particular tables. table_pattern: allow: [.*] # Optional profiling: enabled: true profile_table_level_only: true sink: # config sinks参数说明基础连接参数参数类型默认值说明contact_pointstrlocalhostCassandra 实例的域名或 IP 地址不含端口。portint9042连接 Cassandra 实例的端口号。usernamestr可选None对system_schemakeyspace 具有读取权限的用户名。passwordstr可选密钥类型None与 username 关联的密码配置中会被脱敏处理TransparentSecretStr。SSL 配置可选参数类型默认值说明ssl_ca_certsstr可选NoneCA 证书文件路径用于 SSL 连接。ssl_certfilestr可选NoneSSL 客户端证书文件路径。ssl_keyfilestr可选NoneSSL 客户端私钥文件路径。ssl_versionstr可选TLS_CLIENTSSL 协议版本可选值TLS_CLIENT、TLSv1、TLSv1_1、TLSv1_2、TLSv1_3。底层实现中见 cassandra_api.pySSL 版本字符串会映射到 Pythonssl模块常量TLS_CLIENT→PROTOCOL_TLS_CLIENTTLSv1_2→PROTOCOL_TLSv1_2等同时会校验ssl_certfile与ssl_keyfile必须成对提供只提供一个会抛出ValueError。相关行为有单元测试覆盖见 test_cassandra_source.py 中的test_authenticate_ssl_ca_certs与test_authenticate_ssl_all_certs。Astra DB 云配置cloud_config参数类型默认值说明tokenstr密钥类型必填Astra DB 应用令牌Application Token用于认证。secure_connect_bundlestr必填Secure Connect Bundle.zip的文件路径用于安全连接到 DataStax Astra DB。connect_timeoutint600建立新连接的超时时间秒。request_timeoutint600单个 Cassandra 请求的超时时间秒。使用云配置时源码会构造Cluster(cloud{...})并以固定用户名token 应用令牌作为PlainTextAuthProvider同时强制ProtocolVersion.V4见 cassandra_api.py。这也解释了为什么文档特别提示Astra DB 场景请使用 CLI 摄取元数据。过滤与范围控制参数类型默认值说明keyspace_patternAllowDenyPattern全部允许用于过滤要摄取的 keyspace 的正则模式。table_patternAllowDenyPattern全部允许用于过滤要摄取的keyspace.table的正则模式。profile_patternAllowDenyPattern全部允许用于过滤要剖析的表的正则模式。keyspace_pattern与table_pattern在摄取循环中分别被调用keyspace 级过滤发生在 cassandra.py表/视图级过滤发生在 cassandra.py 与 cassandra.py。注意视图也复用table_pattern进行过滤源码中留有 TODO 注释未来可能引入独立的view_pattern。状态化摄取与删除检测参数类型说明stateful_ingestionStatefulStaleMetadataRemovalConfig状态化摄取与过期元数据移除配置删除检测能力默认开启。Profiling 配置参数类型默认值说明profiling.enabledboolFalse是否启用数据剖析。profiling.profile_table_level_onlybool-仅做表级剖析行数、列数不逐列采样统计。profiling.max_workersint-剖析使用的线程池大小并发剖析多个表。profiling.include_field_null_count等bool-控制字段级统计项null 计数、distinct、min/max、mean、median、stdev、quantiles、sample values。is_profiling_enabled()的逻辑定义在 cassandra_config.py只有profiling.enabledTrue且底层操作配置允许时才执行剖析。摄取流程与底层实现原理CassandraSource的主流程在 get_workunits_internal整体调用链如下CassandraSource.get_workunits_internal() ├── CassandraAPI.authenticate() # 建立连接云/本地 可选 SSL ├── CassandraAPI.get_keyspaces() # 读取 system_schema.keyspaces │ └── 跳过 SYSTEM_KEYSPACE_LIST 中的系统 keyspace │ └── keyspace_pattern 过滤 │ └── 生成 Keyspace Container ├── _extract_tables_from_keyspace() # 每 keyspace 读取表 列 │ ├── CassandraAPI.get_tables() # system_schema.tables │ └── CassandraAPI.get_columns() # system_schema.columns → SchemaField ├── _extract_views_from_keyspace() # 每 keyspace 读取物化视图 │ ├── CassandraAPI.get_views() # system_schema.views │ └── 生成视图 Dataset 与基表的血缘 └── 若 is_profiling_enabled() → CassandraProfiler.get_workunits()系统 keyspace 的自动排除源码中定义了内置的系统 keyspace 集合见 cassandra_utils.pySYSTEM_KEYSPACE_LIST set( [system, system_auth, system_schema, system_distributed, system_traces] )这些系统 keyspace 永远会被跳过不会出现在摄取结果中避免向 DataHub 灌入无业务价值的系统元数据。表级元数据与自定义属性每个表 Dataset 会携带来自system_schema.tables的大量存储配置作为自定义属性见 cassandra.pybloom_filter_fp_chance、caching、compaction、compression、crc_check_chance、dclocal_read_repair_chance、default_time_to_live、extensions、gc_grace_seconds、max_index_interval、min_index_interval、memtable_flush_period_in_ms、read_repair_chance、speculative_retry视图还会额外带上include_all_columns。这些属性对 DBA 在 DataHub 中直接查看表的存储策略很有价值。Schema 类型映射Cassandra 的 CQL 类型会被转换为 DataHub 的SchemaFieldDataType映射表定义在 cassandra_utils.pyCQL 类型DataHub SchemaFieldDataTypebooleanBooleanTypeClassblobBytesTypeClassbigint,counter,decimal,double,float,int,smallint,tinyint,varintNumberTypeClassdateDateTypeClassduration,time,timestampTimeTypeClasstext,ascii,inet,timeuuid,uuid,varcharStringTypeClassgeo_pointRecordTypeClasshistogramArrayTypeClass未知类型NullTypeClass并记录 warning 日志转换器还会记录每个字段的nativeDataType原始 CQL 类型并默认nullableTrue。该逻辑有单元测试覆盖见 test_cassandra_source.py 的test_cassandra_schema_conversion覆盖 timestamp/text 等类型的列路径唯一性。物化视图摄取与血缘Cassandra 物化视图Materialized View会被摄取为Datasetsubtype 为 VIEW并额外输出两部分信息视图逻辑以ViewPropertiesClass形式记录viewLogic取自视图的where_clauseviewLanguage固定为CQLmaterializedTrue见 cassandra.py血缘基于base_table_name生成到基表的UpstreamLineagelineage type 为DatasetLineageTypeClass.VIEW并通过 get_upstream_fields_of_field_in_datasource 生成字段级血缘——视图列与基表同名列之间建立FIELD到FIELD_SET的细粒度血缘。由于物化视图与基表必然处于同一 keyspace源码注释明确指出无需使用base_table_id直接以{keyspace}.{base_table_name}构造上游 URN。Profiling 数据剖析能力启用 profiling 后CassandraProfiler会对每个表生成DatasetProfile工作单元见 cassandra_profiling.py其过程包括行数统计执行SELECT COUNT(*)获取row_count列数统计直接取已摄取列的数目作为column_count字段级剖析当profile_table_level_onlyFalse时对每列统计null_count、distinct_countuniqueCount、min/max、数值列额外计算mean、median、stdev、25%/75% 分位数quantiles并抽样前 5 个样本值sampleValues并发执行通过ThreadPoolExecutor(max_workersprofiling.max_workers)并行剖析多个表类型跳过timeuuid、blob、frozentupletinyint, text等列类型会被跳过SortedSet、OrderedMapSerializedKey、list 等容器类型会被展开后再计算。因此强烈建议在配置中对大表设置采样行数上限并优先使用profile_table_level_only: true只做表级剖析避免对全表数据做字段统计造成性能问题这也正是原文档中 caution 提示的用意。能力、限制与故障排查能力Capabilities以本文“能力表”为准含各能力的默认开启状态所有能力的详细说明以模块文档为准容器Container默认开启Schema 元数据默认开启平台实例Platform Instance默认开启删除检测Deletion Detection默认通过状态化摄取开启。限制Limitations模块行为受限于数据源 API、权限以及平台暴露的元数据。以下情况需要额外注意具体概念映射仍在完善中原文档明确标注 pending视图过滤复用table_pattern暂未提供独立的视图过滤模式未知 CQL 类型会映射为NullTypeClass未启用 Schema 提取或来源不支持时SchemaField 与血缘不会被产出profiling 对数据表有SELECT权限要求且未设置采样上限时可能造成资源压力。故障排查Troubleshooting如果摄取失败请按以下顺序排查凭据用户名/密码或 Astra DB Token 是否正确、是否过期权限是否对system_schema及目标 keyspace 具备SELECT权限启用 profiling 时还需数据表读权限连通性本地部署检查contact_point与port是否可达Astra DB 检查 Secure Connect Bundle 路径是否正确范围过滤确认keyspace_pattern/table_pattern没有误杀目标对象被过滤的对象会出现在 Source Report 的filtered列表中源码见 cassandra_utils.py日志查看摄取日志中的 source 相关错误信息CassandraSourceReport会记录失败的表/视图数量与具体异常据此调整配置。安装与运行方式安装依赖Cassandra Source 属于 metadata-ingestion 的 extras 依赖安装声明见 setup.pycassandraextra 依赖cassandra-driver3.30.1,4.0.0可通过以下方式安装pip install acryl-datahub[cassandra]运行摄取将上文完整 Recipe 保存为cassandra_recipe.yml后执行datahub ingest -c cassandra_recipe.yml运行结束后可在 DataHub 中查看keyspace 容器 → 表/视图 Dataset → Schema 字段 → 血缘以及若启用表/字段级 Profile。小结DataHub 的 Cassandra 集成将 Cassandra/DataStax Astra DB 的 keyspace、表、物化视图与列结构系统地映射到 DataHub 的 Container、Dataset 与 SchemaField 体系并提供状态化删除检测与可选的字段级剖析。从源码看其实现围绕system_schema系统表展开通过 CassandraAPI 封装查询、CassandraSource 组织工作单元、CassandraProfiler 输出剖析结果整体链路清晰、可配置项完备。集成测试样本可参考 tests/integration/cassandra 目录下的 Golden 文件与测试用例便于进一步验证与理解输出格式。【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
DataHub Cassandra 数据源接入指南Keyspace/Table/View 元数据摄取与配置详解【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahubDataHub 官方提供的 Cassandra 集成用于将 Cassandra 集群中的核心元数据数据集/表/视图、Schema 字段、容器摄取进 DataHub并支持基于状态化摄取Stateful Ingestion的删除检测。本文以仓库中的 Cassandra Source 文档 为主线结合 cassandra_pre.md、cassandra_post.md 与完整 示例 Recipe并深入 源码 讲解其原理帮助你完成 Cassandra含 DataStax Astra DB到 DataHub 的元数据同步。Cassandra 集成概览Cassandra 是一个用于存储和查询分析型analytical或操作型operational数据的分布式数据平台。DataHub 的 Cassandra 集成覆盖以下核心元数据实体数据集 / 表 / 视图Dataset/Table/ViewSchema 字段SchemaField容器Container状态化删除检测Stateful Deletion Detection从源码装饰器可以确认该 Source 的能力声明见 cassandra.py能力SourceCapability声明CONTAINERS容器Enabled by defaultSCHEMA_METADATASchema 元数据Enabled by defaultPLATFORM_INSTANCE平台实例Enabled by defaultDELETION_DETECTION删除检测Enabled by default via stateful ingestion该 Source 的 SupportStatus 为BETA用于生产摄取工作流相关模块级能力请以后文及官方文档为准。概念映射Concept Mapping原文档明确指出Cassandra 专属的概念映射细节仍待完善the specific concept mapping is still pending以下是 DataHub 中通用的概念映射关系同样适用于 Cassandra源概念Source ConceptDataHub 概念说明Platform/account/project scopePlatform Instance, Container在平台上下文内组织资产。Core technical asset例如 table/view/topic/fileDataset主要摄取的资产类型。Schema fields / columnsSchemaField支持 Schema 提取时包含。Ownership and collaboration principalsCorpUser, CorpGroup由支持所有权与身份元数据的模块发出。Dependencies and processing relationshipsLineage edges支持并启用了血缘提取时可用。结合源码实现Cassandra 集成实际落地了上述映射的绝大部分具体如下Keyspace → Container在 cassandra.py 中每个 keyspace 会被生成一个Container其subtype为DatasetContainerSubTypes.KEYSPACE并附带durable_writes与replication两个额外属性extra_properties。Table / View → Dataset表被生成为DatasetSubTypes.TABLE类型的数据集物化视图被生成为DatasetSubTypes.VIEW类型的数据集详见 cassandra.py 与 cassandra.py。Column → SchemaField每一列通过CassandraToSchemaFieldConverter转换成SchemaField见 cassandra_utils.py。物化视图 → Lineage edges视图会基于其base_table_name生成与基表之间的血缘边并附带字段级fine-grained血缘这在大多数通用 Source 中并不常见是 Cassandra 集成的亮点之一见下文“视图血缘”章节。摄取前置条件与权限准备在运行摄取之前需要确保到数据源的网络连通性有效的认证凭据该模块所需元数据 API 的读取权限。本模块支持DataStax Astra DB与Cassandra Enterprise EditionEE。获取连接所需信息Astra DB云登录 Astra DB Console进入Organization Settings Token Management生成具有读取所需权限的Application Token从 Astra DB Console 下载Secure Connect Bundle。Cassandra EE本地/自建准备对所需 keyspaces 具有读权限的用户名username与密码password。权限要求用户或 Token 必须具备SELECT权限以支持访问系统 keyspace如system_schema中的元数据获取 keyspaces、tables、columns、views 的信息若启用了数据剖析data profiling还需对数据表执行SELECT操作。源码中实际使用的系统查询均针对system_schema定义在 cassandra_api.py-- 获取所有 keyspace SELECT * FROM system_schema.keyspaces -- 获取某 keyspace 下所有表 SELECT * FROM system_schema.tables WHERE keyspace_name %s -- 获取某表的全部列 SELECT * FROM system_schema.columns WHERE keyspace_name %s AND table_name %s -- 获取某 keyspace 下所有视图 SELECT * FROM system_schema.views WHERE keyspace_name %s -- 行数统计profiling 用 SELECT COUNT(*) AS row_count FROM {keyspace}.{table}验证数据库访问Astra DB确保 Secure Connect Bundle 被正确配置与使用Cassandra 开源版确保contact point与port可访问。:::caution 注意 启用 profiling 时务必对采样的行数设置上限。对超大表不加限制地进行剖析可能导致资源过度消耗与性能下降。 ::::::note 提示 对于 Astra DB 云配置必须在配置中指定 Secure Connect Bundle 路径因此请使用 CLI 方式向 DataHub 摄取元数据。 :::配置详解与完整 RecipeCassandra Source 的配置类定义在 cassandra_config.py下面结合示例 Recipe cassandra_recipe.yml 逐一说明。完整配置示例可直接运行source: type: cassandra config: # Credentials for on prem cassandra contact_point: localhost port: 9042 username: admin password: password # SSL Configuration (optional) #ssl_ca_certs: /path/to/ca-certificate.pem #ssl_certfile: /path/to/client-certificate.pem #ssl_keyfile: /path/to/client-private-key.pem #ssl_version: TLS_CLIENT # Options: TLS_CLIENT, TLSv1, TLSv1_1, TLSv1_2, TLSv1_3 # Or # Credentials Astra Cloud #cloud_config: # secure_connect_bundle: Path to Secure Connect Bundle (.zip) # token: Application Token # Optional Allow / Deny extraction of particular keyspaces. keyspace_pattern: allow: [.*] # Optional Allow / Deny extraction of particular tables. table_pattern: allow: [.*] # Optional profiling: enabled: true profile_table_level_only: true sink: # config sinks参数说明基础连接参数参数类型默认值说明contact_pointstrlocalhostCassandra 实例的域名或 IP 地址不含端口。portint9042连接 Cassandra 实例的端口号。usernamestr可选None对system_schemakeyspace 具有读取权限的用户名。passwordstr可选密钥类型None与 username 关联的密码配置中会被脱敏处理TransparentSecretStr。SSL 配置可选参数类型默认值说明ssl_ca_certsstr可选NoneCA 证书文件路径用于 SSL 连接。ssl_certfilestr可选NoneSSL 客户端证书文件路径。ssl_keyfilestr可选NoneSSL 客户端私钥文件路径。ssl_versionstr可选TLS_CLIENTSSL 协议版本可选值TLS_CLIENT、TLSv1、TLSv1_1、TLSv1_2、TLSv1_3。底层实现中见 cassandra_api.pySSL 版本字符串会映射到 Pythonssl模块常量TLS_CLIENT→PROTOCOL_TLS_CLIENTTLSv1_2→PROTOCOL_TLSv1_2等同时会校验ssl_certfile与ssl_keyfile必须成对提供只提供一个会抛出ValueError。相关行为有单元测试覆盖见 test_cassandra_source.py 中的test_authenticate_ssl_ca_certs与test_authenticate_ssl_all_certs。Astra DB 云配置cloud_config参数类型默认值说明tokenstr密钥类型必填Astra DB 应用令牌Application Token用于认证。secure_connect_bundlestr必填Secure Connect Bundle.zip的文件路径用于安全连接到 DataStax Astra DB。connect_timeoutint600建立新连接的超时时间秒。request_timeoutint600单个 Cassandra 请求的超时时间秒。使用云配置时源码会构造Cluster(cloud{...})并以固定用户名token 应用令牌作为PlainTextAuthProvider同时强制ProtocolVersion.V4见 cassandra_api.py。这也解释了为什么文档特别提示Astra DB 场景请使用 CLI 摄取元数据。过滤与范围控制参数类型默认值说明keyspace_patternAllowDenyPattern全部允许用于过滤要摄取的 keyspace 的正则模式。table_patternAllowDenyPattern全部允许用于过滤要摄取的keyspace.table的正则模式。profile_patternAllowDenyPattern全部允许用于过滤要剖析的表的正则模式。keyspace_pattern与table_pattern在摄取循环中分别被调用keyspace 级过滤发生在 cassandra.py表/视图级过滤发生在 cassandra.py 与 cassandra.py。注意视图也复用table_pattern进行过滤源码中留有 TODO 注释未来可能引入独立的view_pattern。状态化摄取与删除检测参数类型说明stateful_ingestionStatefulStaleMetadataRemovalConfig状态化摄取与过期元数据移除配置删除检测能力默认开启。Profiling 配置参数类型默认值说明profiling.enabledboolFalse是否启用数据剖析。profiling.profile_table_level_onlybool-仅做表级剖析行数、列数不逐列采样统计。profiling.max_workersint-剖析使用的线程池大小并发剖析多个表。profiling.include_field_null_count等bool-控制字段级统计项null 计数、distinct、min/max、mean、median、stdev、quantiles、sample values。is_profiling_enabled()的逻辑定义在 cassandra_config.py只有profiling.enabledTrue且底层操作配置允许时才执行剖析。摄取流程与底层实现原理CassandraSource的主流程在 get_workunits_internal整体调用链如下CassandraSource.get_workunits_internal() ├── CassandraAPI.authenticate() # 建立连接云/本地 可选 SSL ├── CassandraAPI.get_keyspaces() # 读取 system_schema.keyspaces │ └── 跳过 SYSTEM_KEYSPACE_LIST 中的系统 keyspace │ └── keyspace_pattern 过滤 │ └── 生成 Keyspace Container ├── _extract_tables_from_keyspace() # 每 keyspace 读取表 列 │ ├── CassandraAPI.get_tables() # system_schema.tables │ └── CassandraAPI.get_columns() # system_schema.columns → SchemaField ├── _extract_views_from_keyspace() # 每 keyspace 读取物化视图 │ ├── CassandraAPI.get_views() # system_schema.views │ └── 生成视图 Dataset 与基表的血缘 └── 若 is_profiling_enabled() → CassandraProfiler.get_workunits()系统 keyspace 的自动排除源码中定义了内置的系统 keyspace 集合见 cassandra_utils.pySYSTEM_KEYSPACE_LIST set( [system, system_auth, system_schema, system_distributed, system_traces] )这些系统 keyspace 永远会被跳过不会出现在摄取结果中避免向 DataHub 灌入无业务价值的系统元数据。表级元数据与自定义属性每个表 Dataset 会携带来自system_schema.tables的大量存储配置作为自定义属性见 cassandra.pybloom_filter_fp_chance、caching、compaction、compression、crc_check_chance、dclocal_read_repair_chance、default_time_to_live、extensions、gc_grace_seconds、max_index_interval、min_index_interval、memtable_flush_period_in_ms、read_repair_chance、speculative_retry视图还会额外带上include_all_columns。这些属性对 DBA 在 DataHub 中直接查看表的存储策略很有价值。Schema 类型映射Cassandra 的 CQL 类型会被转换为 DataHub 的SchemaFieldDataType映射表定义在 cassandra_utils.pyCQL 类型DataHub SchemaFieldDataTypebooleanBooleanTypeClassblobBytesTypeClassbigint,counter,decimal,double,float,int,smallint,tinyint,varintNumberTypeClassdateDateTypeClassduration,time,timestampTimeTypeClasstext,ascii,inet,timeuuid,uuid,varcharStringTypeClassgeo_pointRecordTypeClasshistogramArrayTypeClass未知类型NullTypeClass并记录 warning 日志转换器还会记录每个字段的nativeDataType原始 CQL 类型并默认nullableTrue。该逻辑有单元测试覆盖见 test_cassandra_source.py 的test_cassandra_schema_conversion覆盖 timestamp/text 等类型的列路径唯一性。物化视图摄取与血缘Cassandra 物化视图Materialized View会被摄取为Datasetsubtype 为 VIEW并额外输出两部分信息视图逻辑以ViewPropertiesClass形式记录viewLogic取自视图的where_clauseviewLanguage固定为CQLmaterializedTrue见 cassandra.py血缘基于base_table_name生成到基表的UpstreamLineagelineage type 为DatasetLineageTypeClass.VIEW并通过 get_upstream_fields_of_field_in_datasource 生成字段级血缘——视图列与基表同名列之间建立FIELD到FIELD_SET的细粒度血缘。由于物化视图与基表必然处于同一 keyspace源码注释明确指出无需使用base_table_id直接以{keyspace}.{base_table_name}构造上游 URN。Profiling 数据剖析能力启用 profiling 后CassandraProfiler会对每个表生成DatasetProfile工作单元见 cassandra_profiling.py其过程包括行数统计执行SELECT COUNT(*)获取row_count列数统计直接取已摄取列的数目作为column_count字段级剖析当profile_table_level_onlyFalse时对每列统计null_count、distinct_countuniqueCount、min/max、数值列额外计算mean、median、stdev、25%/75% 分位数quantiles并抽样前 5 个样本值sampleValues并发执行通过ThreadPoolExecutor(max_workersprofiling.max_workers)并行剖析多个表类型跳过timeuuid、blob、frozentupletinyint, text等列类型会被跳过SortedSet、OrderedMapSerializedKey、list 等容器类型会被展开后再计算。因此强烈建议在配置中对大表设置采样行数上限并优先使用profile_table_level_only: true只做表级剖析避免对全表数据做字段统计造成性能问题这也正是原文档中 caution 提示的用意。能力、限制与故障排查能力Capabilities以本文“能力表”为准含各能力的默认开启状态所有能力的详细说明以模块文档为准容器Container默认开启Schema 元数据默认开启平台实例Platform Instance默认开启删除检测Deletion Detection默认通过状态化摄取开启。限制Limitations模块行为受限于数据源 API、权限以及平台暴露的元数据。以下情况需要额外注意具体概念映射仍在完善中原文档明确标注 pending视图过滤复用table_pattern暂未提供独立的视图过滤模式未知 CQL 类型会映射为NullTypeClass未启用 Schema 提取或来源不支持时SchemaField 与血缘不会被产出profiling 对数据表有SELECT权限要求且未设置采样上限时可能造成资源压力。故障排查Troubleshooting如果摄取失败请按以下顺序排查凭据用户名/密码或 Astra DB Token 是否正确、是否过期权限是否对system_schema及目标 keyspace 具备SELECT权限启用 profiling 时还需数据表读权限连通性本地部署检查contact_point与port是否可达Astra DB 检查 Secure Connect Bundle 路径是否正确范围过滤确认keyspace_pattern/table_pattern没有误杀目标对象被过滤的对象会出现在 Source Report 的filtered列表中源码见 cassandra_utils.py日志查看摄取日志中的 source 相关错误信息CassandraSourceReport会记录失败的表/视图数量与具体异常据此调整配置。安装与运行方式安装依赖Cassandra Source 属于 metadata-ingestion 的 extras 依赖安装声明见 setup.pycassandraextra 依赖cassandra-driver3.30.1,4.0.0可通过以下方式安装pip install acryl-datahub[cassandra]运行摄取将上文完整 Recipe 保存为cassandra_recipe.yml后执行datahub ingest -c cassandra_recipe.yml运行结束后可在 DataHub 中查看keyspace 容器 → 表/视图 Dataset → Schema 字段 → 血缘以及若启用表/字段级 Profile。小结DataHub 的 Cassandra 集成将 Cassandra/DataStax Astra DB 的 keyspace、表、物化视图与列结构系统地映射到 DataHub 的 Container、Dataset 与 SchemaField 体系并提供状态化删除检测与可选的字段级剖析。从源码看其实现围绕system_schema系统表展开通过 CassandraAPI 封装查询、CassandraSource 组织工作单元、CassandraProfiler 输出剖析结果整体链路清晰、可配置项完备。集成测试样本可参考 tests/integration/cassandra 目录下的 Golden 文件与测试用例便于进一步验证与理解输出格式。【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
