企业官网建设流程全解析

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 展示了两个典型变体：

1. sql_ci变体（sql_ci_no_collision）：以小概率（用户可配置）生成碰撞，通过在名称上追加确定性标签避免碰撞；
2. sql_cs变体（sql_cs_latin2）：允许 LATIN（即 umlaut）字符的大小写敏感命名约定。

自定义命名约定的推荐模块布局

自定义命名约定是从dlt.common.normalizers.naming导入的NamingConvention基类派生的类。官方推荐以下模块布局：

1. 每个命名约定独立放在一个 Python 模块（文件）中；
2. 类名统一命名为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）：

类属性PATH_SEPARATOR（"__"）决定嵌套字段与表名的连接符，_DEFAULT_COLLISION_PROB（0.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_convention="sql_cs_latin2") pipeline = dlt.pipeline( pipeline_name="sql_cs_latin2_pipeline", destination=dest_, dataset_name="example_data", dev_mode=True, ) # 提取、归一化并加载数据 load_info = pipeline.run([{"StückId": 1}], table_name="Ausrü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_convention="sql_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] naming="sql_ci_no_collision"

对应的管线代码：

# sql_ci_no_collision（在 config.toml 中配置） # 注意：名为 `sql_ci_no_collision` 的管线将创建同名默认 schema， # 因此我们可以在 config.toml 中利用这一名称，只影响这条管线，而不动上面的 postgres 管线 pipeline = dlt.pipeline( pipeline_name="sql_ci_no_collision", destination="duckdb", dataset_name="example_data", dev_mode=True, ) # duckdb 大小写不敏感，下面的表和列本会冲突，sql_ci_no_collision 避免了这一点 data_1 = {"ItemID": 1, "itemid": "collides"} load_info = pipeline.run([data_1], table_name="BigData") data_2 = {"1Data": 1, "_1data": "collides"} # 使用会碰撞的表名 load_info = pipeline.run([data_2], table_name="bigdata")

配置生效的关键机制（详见 naming-convention.md）：

• [schema] naming适用于所有管线（schema）；环境变量SCHEMA__NAMING效果相同；
• 也可以按 source 细分配置：[sources.zendesk.schema] naming="sql_cs_v1"；
• 显式配置覆盖一切其他设置：既改变已创建 schema 中存储的命名约定，也覆盖目标能力中的偏好；
• 由于 dlt 用管线名创建默认 schema 名，示例巧妙地让config.toml的全局配置只作用于同名的这条管线。

用命名约定反查表名

因为标签是确定性的，示例直接用命名约定本身来获取实际表名，随后执行DESCRIBE TABLE：

with 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 = f"DESCRIBE TABLE {first_table}" print(sql) print(conn.sql(sql)) second_table = pipeline.default_schema.naming.normalize_table_identifier("bigdata") sql = f"DESCRIBE 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 目录内置了以下约定，可作为自定义时的对照参考：

自定义约定的完整配置示例同样被仓库测试使用：测试目录 tests/common/cases/normalizers 中的sql_upper.py、title_case.py、snake_no_x.py都是按"独立模块 +NamingConvention类名"布局编写的自定义约定，并被命名约定测试套件（test_naming.py 等）实际加载验证，例如：

[schema] naming="tests.common.cases.normalizers.sql_upper"

dlt 将导入该模块并使用其中的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.py），dlt 决定是否为标识符加引号保留大小写。

因此自定义约定必须如实声明：若你的归一化结果保留大小写且希望被引号引用，is_case_sensitive应返回True；若输出全部小写、不区分大小写，则应返回False（如sql_ci_v1所做的那样）。

标识符缩短机制（超长标识符）

目标库对标识符长度有限制，dlt 在归一化阶段从目标能力中取得max_identifier_length并截断超长标识符。基类静态方法shorten_identifier（naming.py）的逻辑是：

1. 若归一化后长度超过max_length，用原始标识符计算确定性哈希标签；
2. 把标签插入截断字符串的中间（_trim_and_tag保留前后各一半），这样即使被截断，标识符仍以高概率保持唯一；
3. 所有内置与自定义约定在normalize_identifier末尾都会调用self.shorten_identifier(norm_identifier, identifier, self.max_length)完成截断。

因此自定义约定同样受益于该机制，无需自己实现截断逻辑。

最佳实践与注意事项

• 保持统一命名约定：dlt 默认对所有目标库使用同一命名约定，除非你明确指定，否则不要轻易为不同目标混用不同约定，以免同一批数据在不同库中表列名不一致。
• 避免传模块对象：显式指定时传模块名字符串（如naming_convention="my_package.sql_cs_latin2"），不要import后传模块对象，否则并行归一化时可能遇到 pickle 错误。
• 注意约定变更的破坏性：schema 中保存命名约定的全限定名，加载 schema 时 dlt 会尝试导入它；若改变命名约定导致已存在表的标识符发生变化，归一化会失败以阻止意外的 schema 迁移。因此自定义约定需要随管线代码或 pip 包一起分发。
• dataset_name 归一化可单独关闭：[destination.snowflake] enable_dataset_name_normalization=false可让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