dlt 自定义命名约定Naming Convention实战从零实现防碰撞与支持拉丁字符的标识符翻译器【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt数据加载工具 dlt 会把源数据中的任意 Unicode 标识符翻译为目标数据库允许的合法标识符这一翻译规则由命名约定naming convention定义。本文基于 dlt 官方示例 custom_naming.md 与仓库源码完整讲解如何编写、注册和运行自定义命名约定你将掌握推荐模块布局、覆盖is_case_sensitive与normalize_identifier两种扩展点并能在postgres目标工厂与config.toml两种方式下启用自己的命名约定最终实现永不碰撞与允许拉丁字符两类实用场景。为什么需要自定义命名约定dlt 从 JSON 流等数据源中读取字段名、表名并生成目标库标识符。源数据中的键名可以是任意 Unicode 字符、任意长度、任意命名风格如StückId、BigData而目标库Postgres、Redshift、DuckDB 等对合法标识符有严格限制例如 Redshift 只接受最长 127 个字符的、大小写不敏感的字母数字标识符。命名约定本质上是一组把源标识符字符串映射为目标标识符字符串的函数。默认情况下 dlt 对所有目标库使用同一套命名约定默认是snake_case保证用户在不同数据库中看到的表名和列名一致。当你需要允许源数据中的 UNICODE 字母如德语 umlaut 字符ü原样进入目标库在大小写不敏感的目标库中避免ItemID与itemid这类归一化后碰撞的标识符使用与内置约定不同的字符折叠规则例如全大写、保留column_前缀等就需要编写自定义命名约定。仓库中的官方示例 custom_naming.md 展示了两个典型变体sql_ci变体sql_ci_no_collision以小概率用户可配置生成碰撞通过在名称上追加确定性标签避免碰撞sql_cs变体sql_cs_latin2允许 LATIN即 umlaut字符的大小写敏感命名约定。自定义命名约定的推荐模块布局自定义命名约定是从dlt.common.normalizers.naming导入的NamingConvention基类派生的类。官方推荐以下模块布局每个命名约定独立放在一个 Python 模块文件中类名统一命名为NamingConvention。这样 dlt 就可以通过完全限定模块名来定位约定导入该模块后直接使用其中的NamingConvention类。示例中sql_cs_latin2与sql_ci_no_collision即是两个可导入的模块名它们需要按上述布局定义在你自己的包中例如my_package/sql_cs_latin2.py与my_package/sql_ci_no_collision.py示例代码通过# import sql_cs_latin2 # is resolving in this context说明其在当前上下文中可解析。基类定义在 naming.py它声明了两个必须由子类实现的核心成员class NamingConvention(ABC): def __init__(self, max_length: int None) - None: self.max_length max_length property abstractmethod def is_case_sensitive(self) - bool: 告诉 dlt 该约定产生大小写敏感还是不敏感的标识符 pass abstractmethod def normalize_identifier(self, identifier: str) - str: 按照本约定的规则归一化并缩短标识符 if identifier is None: raise ValueError(name is None) identifier identifier.strip() if not identifier: raise ValueError(identifier) return identifier基类还提供了一组可直接复用的方法见 naming.py方法作用normalize_table_identifier归一化将作为 dataset、table 或 schema 名的标识符默认委托给normalize_identifiermake_path/break_path用PATH_SEPARATOR默认__拼接/拆分嵌套字段与表名路径normalize_path/normalize_tables_path逐段归一化路径组件后整体缩短shorten_identifier静态方法当标识符超长时计算确定性标签并截断到max_length_compute_tag基于hashlib.shake_128计算确定性标签碰撞概率由collision_prob参数控制_remove_leading_underscores移除开头的下划线类属性PATH_SEPARATOR__决定嵌套字段与表名的连接符_DEFAULT_COLLISION_PROB0.001是默认的标签碰撞概率。场景一允许拉丁字符的大小写敏感约定sql_cs_latin2内置的sql_cs_v1源码见 sql_cs_v1.py生成 SQL 安全的、保留源大小写的标识符但它通过正则[^a-zA-Z\d_]丢弃所有非 ASCII 字母数字字符——StückId中的ü会被替换为下划线。当目标库如 Postgres本身接受 UNICODE 字母时我们更希望保留这些字符。基于源码模式可以推断sql_cs_latin2的实现思路是继承sql_cs_v1或直接继承基类把清洗规则中仅保留 ASCII 字母数字放宽为保留包括拉丁字母在内的 UNICODE 字母同时维持大小写敏感语义# my_package/sql_cs_latin2.py import re from dlt.common.normalizers.naming.naming import NamingConvention as BaseNamingConvention RE_NON_ALPHANUMERIC re.compile(r[^\w\d_]) # \w 匹配 UNICODE 字母数字 class NamingConvention(BaseNamingConvention): property def is_case_sensitive(self) - bool: return True # 覆盖基类默认值声明本约定大小写敏感 def normalize_identifier(self, identifier: str) - str: identifier super().normalize_identifier(identifier) norm_identifier RE_NON_ALPHANUMERIC.sub(_, identifier) return self.shorten_identifier(norm_identifier, identifier, self.max_length)仓库中的 sql_upper.py 测试案例印证了这一覆盖模式它重写is_case_sensitive返回True并在normalize_identifier中用str.translate清洗字符后调用shorten_identifier完成截断。此外内置的duck_case见 duck_case.py也展示了保留所有 Unicode 字符除换行的宽松策略可作为参考。通过目标工厂显式指定使用自定义约定最直接的方式是在创建目标时传入模块名import dlt # sql_cs_latin2 是一个可导入的模块名 dest_ dlt.destinations.postgres(naming_conventionsql_cs_latin2) pipeline dlt.pipeline( pipeline_namesql_cs_latin2_pipeline, destinationdest_, dataset_nameexample_data, dev_modeTrue, ) # 提取、归一化并加载数据 load_info pipeline.run([{StückId: 1}], table_nameAusrüstung) print(load_info) with pipeline.sql_client() as client: # 注意大小写敏感标识符需要加引号 with client.execute_query(SELECT StückId FROM Ausrüstung) as cur: print(cur.description) print(cur.fetchone())关键点dlt.destinations.postgres(naming_conventionsql_cs_latin2)把约定挂在目标工厂上它会覆盖目标自身的首选约定并成为整个管线的默认约定参见 capabilities.py 中generic_capabilities对naming_convention的合并逻辑caps.naming_convention naming_convention or caps.naming_convention。示例注释指出sql_cs_latin2is case sensitive and postgres accepts UNICODE letters in identifiers。Postgres 支持大小写敏感标识符但查询时必须按 SQL 规则加双引号StückId、Ausrüstung否则会被折叠为小写。场景二永不碰撞的大小写不敏感约定sql_ci_no_collision在大小写不敏感的目标库如 DuckDB、Redshift中源数据里ItemID与itemid归一化后可能变成同一个标识符导致列/表碰撞甚至数据被破坏。sql_ci_no_collision的思路是继承sql_ci大小写不敏感、全部小写化在每次归一化时为标识符追加一个确定性标签使不同的源标识符即使归一化后也保持不同。# my_package/sql_ci_no_collision.py from dlt.common.normalizers.naming.sql_ci_v1 import NamingConvention as SqlCiNamingConvention from dlt.common.normalizers.naming.naming import NamingConvention class NamingConvention(SqlCiNamingConvention): def normalize_identifier(self, identifier: str) - str: # 先按 sql_ci 规则归一化小写、去非法字符 norm super().normalize_identifier(identifier) # 追加确定性标签相同源标识符得到相同标签不同源标识符以低概率碰撞 tag NamingConvention._compute_tag(identifier, NamingConvention._DEFAULT_COLLISION_PROB) return f{norm}_{tag} property def is_case_sensitive(self) - bool: return False这一实现的原理直接来自基类源码naming.py_compute_tag使用hashlib.shake_128对源标识符做可扩展输出哈希再经base64编码并小写化生成位数取决于目标碰撞概率的确定性标签_DEFAULT_COLLISION_PROB默认为0.001即理论碰撞概率低于千分之一。由于标签只依赖源标识符本身同一源标识符永远得到同一目标标识符因此查询时可以用pipeline.default_schema.naming.normalize_table_identifier(BigData)反推出实际表名。通过 config.toml 配置示例的第二段使用config.toml方式启用该约定从而只影响名为sql_ci_no_collision的管线[schema] namingsql_ci_no_collision对应的管线代码# sql_ci_no_collision在 config.toml 中配置 # 注意名为 sql_ci_no_collision 的管线将创建同名默认 schema # 因此我们可以在 config.toml 中利用这一名称只影响这条管线而不动上面的 postgres 管线 pipeline dlt.pipeline( pipeline_namesql_ci_no_collision, destinationduckdb, dataset_nameexample_data, dev_modeTrue, ) # duckdb 大小写不敏感下面的表和列本会冲突sql_ci_no_collision 避免了这一点 data_1 {ItemID: 1, itemid: collides} load_info pipeline.run([data_1], table_nameBigData) data_2 {1Data: 1, _1data: collides} # 使用会碰撞的表名 load_info pipeline.run([data_2], table_namebigdata)配置生效的关键机制详见 naming-convention.md[schema] naming适用于所有管线schema环境变量SCHEMA__NAMING效果相同也可以按 source 细分配置[sources.zendesk.schema] namingsql_cs_v1显式配置覆盖一切其他设置既改变已创建 schema 中存储的命名约定也覆盖目标能力中的偏好由于 dlt 用管线名创建默认 schema 名示例巧妙地让config.toml的全局配置只作用于同名的这条管线。用命名约定反查表名因为标签是确定性的示例直接用命名约定本身来获取实际表名随后执行DESCRIBE TABLEwith pipeline.sql_client() as client: from duckdb import DuckDBPyConnection conn: DuckDBPyConnection client.native_connection first_table pipeline.default_schema.naming.normalize_table_identifier(BigData) sql fDESCRIBE TABLE {first_table} print(sql) print(conn.sql(sql)) second_table pipeline.default_schema.naming.normalize_table_identifier(bigdata) sql fDESCRIBE TABLE {second_table} print(sql) print(conn.sql(sql))pipeline.default_schema.naming即当前 schema 使用的命名约定实例normalize_table_identifier对表名类标识符做归一化默认委托给normalize_identifier因此BigData与bigdata会得到两个不同的表名DuckDB 中即可分别DESCRIBE两张表。命名约定的内置选项与源码佐证仓库 dlt/common/normalizers/naming 目录内置了以下约定可作为自定义时的对照参考约定大小写特点snake_case默认不敏感转为小写 snake_case/*→x、-→_、→a、\|→l数字开头补_尾部_→xsql_cs_v1敏感生成 SQL 安全标识符保留源大小写非 ASCII 字母数字替换为_sql_ci_v1不敏感在sql_cs_v1基础上整体小写化见 sql_ci_v1.pyduck_case敏感允许所有 Unicode 字符包括 emoji__作为路径分隔direct敏感允许所有 Unicode 字符且不收缩连续下划线s3_tables不敏感扩展snake_case以符合 S3 Tables 命名规则表名以dlt_而非_dlt_开头自定义约定的完整配置示例同样被仓库测试使用测试目录 tests/common/cases/normalizers 中的sql_upper.py、title_case.py、snake_no_x.py都是按独立模块 NamingConvention类名布局编写的自定义约定并被命名约定测试套件test_naming.py 等实际加载验证例如[schema] namingtests.common.cases.normalizers.sql_upperdlt 将导入该模块并使用其中的NamingConvention类。测试还验证了标签计算的确定性test_json_relational.py中多次以NamingConvention._compute_tag(identifier, _DEFAULT_COLLISION_PROB)断言不同输入得到稳定标签。覆盖 is_case_sensitive 的意义is_case_sensitive属性不仅是一个描述性标记它直接参与 dlt 的标识符碰撞检测见 naming-convention.md 与 L186-L193在大小写不敏感的目标上使用大小写敏感的命名约定时dlt 会检测到碰撞并在加载前中止防止数据被破坏配合目标能力中的has_case_sensitive_identifiers与casefold_identifier见 capabilities.pydlt 决定是否为标识符加引号保留大小写。因此自定义约定必须如实声明若你的归一化结果保留大小写且希望被引号引用is_case_sensitive应返回True若输出全部小写、不区分大小写则应返回False如sql_ci_v1所做的那样。标识符缩短机制超长标识符目标库对标识符长度有限制dlt 在归一化阶段从目标能力中取得max_identifier_length并截断超长标识符。基类静态方法shorten_identifiernaming.py的逻辑是若归一化后长度超过max_length用原始标识符计算确定性哈希标签把标签插入截断字符串的中间_trim_and_tag保留前后各一半这样即使被截断标识符仍以高概率保持唯一所有内置与自定义约定在normalize_identifier末尾都会调用self.shorten_identifier(norm_identifier, identifier, self.max_length)完成截断。因此自定义约定同样受益于该机制无需自己实现截断逻辑。最佳实践与注意事项保持统一命名约定dlt 默认对所有目标库使用同一命名约定除非你明确指定否则不要轻易为不同目标混用不同约定以免同一批数据在不同库中表列名不一致。避免传模块对象显式指定时传模块名字符串如naming_conventionmy_package.sql_cs_latin2不要import后传模块对象否则并行归一化时可能遇到 pickle 错误。注意约定变更的破坏性schema 中保存命名约定的全限定名加载 schema 时 dlt 会尝试导入它若改变命名约定导致已存在表的标识符发生变化归一化会失败以阻止意外的 schema 迁移。因此自定义约定需要随管线代码或 pip 包一起分发。dataset_name 归一化可单独关闭[destination.snowflake] enable_dataset_name_normalizationfalse可让dataset_name保持原样默认true但对某些目标可能产生非法名称谨慎使用。碰撞检测有边界dlt 能检测大小写敏感约定用在大小写不敏感目标等碰撞但不会在归一化源数据时检测字典键碰撞——如果源字典里两个键归一化后相同它们会被合并。这正是sql_ci_no_collision这类无碰撞约定的用武之地。通过本文的两种自定义约定你可以在保留目标库标识符合法性的同时获得保留拉丁字符或消除碰撞的精确控制——自定义命名约定的全部扩展点is_case_sensitive、normalize_identifier、normalize_table_identifier、PATH_SEPARATOR都可以在基类 naming.py 中找到对应实现结合 naming-convention.md 中的配置说明即可上手。【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考