OpenMetadata SFTP 连接器接入指南从连接配置到文件元数据采集【免费下载链接】OpenMetadataThe Open Context Layer for Data and AI , OpenMetadata is the open platform for building trusted data context and business semantics for humans, AI assistants, and agents.项目地址: https://gitcode.com/GitHub_Trending/op/OpenMetadataSFTPSSH File Transfer Protocol连接器是 OpenMetadata Drive驱动类服务家族的一员用于把 SFTP 服务器上的目录与文件作为可治理的数据资产纳入元数据平台。本文基于仓库中的 SFTP 连接器文档Sftp.md与对应源码实现完整讲解 SFTP 服务的连接配置、认证方式、过滤规则与元数据采集行为读者读完可以独立完成一个 SFTP 服务的创建、测试连接与元数据管道配置。一、SFTP 连接器在 OpenMetadata 中的定位OpenMetadata 通过统一的Drive Service抽象来治理文件型数据源Google Drive、SharePoint、OneDrive、SFTP 等。从源码的拓扑定义看drive_service.py 中DriveServiceTopology描述的层级为service - directory - file / spreadsheet - worksheetSFTP 是该抽象下的一种具体实现其服务规范注册于 service_spec.pyServiceSpec BaseSpec(metadata_source_classSftpSource, connection_classSftpConnection)即 SFTP 使用SftpSource采集器与SftpConnection连接器组合。与 Google Drive 不同SFTP 场景下不涉及 Spreadsheet/Worksheet 实体——在 metadata.py 中get_spreadsheets_list、yield_spreadsheet、yield_worksheet等方法均返回空因此 SFTP 采集只产出Directory目录与File文件两类实体。二、前置要求要采集 SFTP 服务器上的元数据用于登录的账号必须对需要编目的目录和文件具有读取权限。这是文档明确列出的唯一硬性要求——连接器本身只做元数据读取列出目录、读取文件头、提取 CSV/TSV 结构不会对服务器上的数据做任何写操作。从源码角度看SFTP 底层基于paramiko库实现见 connection.py 的import paramiko通过TransportSFTPClient建立通道连接的建立、鉴权、资源关闭都在该模块中完成。三、连接配置详解Connection DetailsSFTP 服务的连接配置在 UI 中以表单形式呈现其字段与默认值由 JSON Schema 定义在 sftpConnection.json。该 Schema 要求host与authType为必填项其余字段均有默认值或可省略。1. Host主机SFTP 服务器的主机名或 IP 地址例如sftp.example.com或192.168.1.100。对应 Schema 中host字段类型为字符串。2. Port端口SFTP 服务器端口默认22。源码中同样以connection.port or 22作为兜底逻辑见 connection.py即未显式配置时使用 22。如果服务器运行在非标准端口如企业内部常用的 2222需在此显式指定。3. Authentication Type认证类型SFTP 支持两种认证方式对应 Schema 中authType的oneOf分支认证方式Schema 定义说明Username/PasswordbasicAuth使用用户名 密码认证username与password均为必填Private KeykeyAuth使用 PEM 格式的 SSH 私钥认证username与privateKey为必填两种方式在源码中的处理路径connection.py密码认证transport.connect(username..., password...)私钥认证先解析私钥为paramiko.PKey再transport.connect(username..., pkeypkey)4. Username用户名SFTP 登录用户名两种认证方式都需要。5. Password密码密码认证使用的口令。Schema 中该字段标记为format: password在 UI 与日志中会作为敏感信息处理。6. Private Key私钥PEM 格式的 SSH 私钥内容用于私钥认证。文档与 Schema 描述支持 RSA、Ed25519、ECDSA 与 DSS 四种密钥类型。需要留意的是从源码实现看connection.py 的_parse_private_key实际依次尝试paramiko.RSAKey、paramiko.Ed25519Key、paramiko.ECDSAKey三种类型进行解析且代码注释明确指出paramiko自 4.0.0 起移除了DSSKeyOpenSSH 自 7.0 起也默认拒绝 ssh-dss。因此在实际使用中推荐使用 RSA、Ed25519 或 ECDSA 私钥解析失败时会抛出 Unable to parse private key. Ensure it is in PEM format 的错误提示。7. Private Key Passphrase私钥口令当私钥文件本身被加密带 passphrase时用于解密私钥。若私钥未加密则留空即可。源码中该值作为from_private_key(key_file, passwordpassphrase)的password参数传入。8. Root Directories根目录要扫描的文件与子目录根路径列表默认值为/即用户家目录。可以配置多个目录将采集范围限定在服务器上的特定路径。例如/扫描用户家目录下的全部内容[/data/landing, /data/export]只扫描这两个目录及其子树在 Schema 中该字段类型为字符串数组默认[/]。源码层面metadata.py 的_fetch_directories会遍历rootDirectories未配置时取[/]递归构建目录树同时记录根目录前缀用于后续 FQN 的路径剥离与重建。9. Connection Options连接选项额外的连接选项用于构建发送给服务的连接 URL。对应 Schema 中connectionOptions字段引用自 connectionBasicType.json 的通用定义。对 SFTP 而言通常无需配置。10. Connection Arguments连接参数额外的连接参数如安全或协议相关的配置会在连接时发送给服务。同样引用自通用的connectionArguments定义属于高级可选项。四、目录与文件过滤SFTP 连接配置中包含两个过滤模式字段用于控制采集范围Directory Filter Pattern目录过滤正则表达式只包含/排除匹配的目录。File Filter Pattern文件过滤正则表达式只包含/排除匹配的文件。两者的结构由 filterPattern.json 定义均为includes与excludes两个正则字符串列表。配置示例{ directoryFilterPattern: { includes: [finance.*], excludes: [tmp, archive.*] }, fileFilterPattern: { includes: [.*\\.csv$, .*\\.tsv$], excludes: [.*\\.log$] } }过滤逻辑在源码中体现为目录filter_by_directory(self.source_config.directoryFilterPattern, ...)metadata.pyget_directory_names文件filter_by_file(self.source_config.fileFilterPattern, file_info.name)yield_file方法中对根目录文件与各目录内文件均执行。此外DriveServiceMetadataPipeline还提供useFqnForFiltering选项默认false开启后正则将作用于完整限定名如service_name.directory_name.file_name而非裸名称适用于需要跨目录精确过滤的场景。五、结构化数据采集开关1. Structured Data Files Only仅结构化数据文件默认false。开启后只编目能够提取出 Schema 的结构化数据文件CSV、TSV图片、PDF、视频等非结构化文件会被跳过。源码中结构化文件的判定集合为metadata.pyCSV_EXTENSIONS {.csv, .tsv} STRUCTURED_DATA_EXTENSIONS {.csv, .tsv, .json, .jsonl, .parquet, .avro}注意区分_is_csv_file只识别.csv/.tsv决定是否做 Schema 提取而_is_structured_data_file使用更宽的扩展名集合决定是否被采集为 File 实体。也就是说关闭该开关时所有文件都会被编目为 File 实体但只有 CSV/TSV 会被提取列 Schema开启该开关时JSON/Parquet/Avro 等文件也会被编目它们属于结构化数据图片/PDF/视频则被跳过。2. Extract Sample Data提取样本数据默认false默认关闭以避免性能开销。开启后会从结构化文件CSV、TSV中提取样本数据写入对应 File 实体便于在 UI 中预览文件内容。源码中的样本提取逻辑_extract_csv_schema与_ingest_sample_data_for_file远程读取文件内容self.client.sftp.open(file_path, r)按 UTF-8 解码使用pandas.read_csv解析CHUNKSIZE 200行作为读取上限列类型通过PANDAS_DTYPE_MAP映射为 OpenMetadata 的DataTypeint64/int32 - INT、float64/float32 - FLOAT、bool - BOOLEAN、datetime64[ns] - DATETIME、object - STRING未匹配到的类型统一降级为STRING样本数据取前MAX_SAMPLE_ROWS 50行构建为TableData后通过metadata.ingest_file_sample_data写入。TSV 文件会自动使用制表符\t作为分隔符_get_csv_separator按扩展名判断。六、采集管道配置DriveServiceMetadataPipelineSFTP 属于 Drive 类服务其元数据管道类型为DriveMetadata配置项定义于 driveServiceMetadataPipeline.json。除上述过滤项外常用配置包括配置项默认值说明includeDirectoriestrue是否采集目录元数据includeFilestrue是否采集文件元数据markDeletedDirectoriestrue源端目录被删除后在 OpenMetadata 中软删除对应目录及其关联实体markDeletedFilestrue源端文件被删除后软删除对应文件includeTagstrue是否采集标签includeOwnersfalse是否将匹配的负责人关联到实体overrideMetadatafalse是否用源端元数据覆盖服务器上已有的描述、标签、负责人等threads1并行采集线程数markDeletedDirectories/markDeletedFiles的实现位于 drive_service.py 的mark_directories_as_deleted/mark_files_as_deleted通过delete_entity_from_source对照本次采集的实体状态集做差集删除。SFTP 实现中重写了register_record_directory与register_record_file确保嵌套目录与根目录文件的 FQN 与入库结果严格一致避免刚采集就被当作陈旧实体删除的问题。一个完整的 SFTP 元数据管道 YAML 骨架如下serviceConnection部分即前文配置source: type: sftp serviceName: my_sftp_server serviceConnection: config: type: Sftp host: sftp.example.com port: 22 authType: username: data_reader password: ${SFTP_PASSWORD} rootDirectories: - /data/landing structuredDataFilesOnly: false extractSampleData: false directoryFilterPattern: excludes: - tmp fileFilterPattern: includes: - .*\.(csv|tsv|json)$ sourceConfig: config: type: DriveMetadata includeDirectories: true includeFiles: true markDeletedFiles: true threads: 2 sink: type: metadata-rest config: {} workflowConfig: openMetadataServerConfig: hostPort: http://localhost:8585/api authProvider: openmetadata七、连接测试与采集流程连接测试Test ConnectionSFTP 连接测试由 connection.py 的test_connection实现包含两个校验步骤CheckAccess执行client.sftp.stat(.)验证能否成功认证并访问服务器ListDirectories遍历每个rootDirectories执行client.sftp.listdir(root_dir)验证目录可列出并统计各目录下的条目数。任一步骤失败都会抛出SourceConnectionExceptionUI 上会明确显示是哪一步、哪个目录失败便于排查权限与路径问题。元数据采集流程采集器SftpSourcemetadata.py的完整流程为建立连接并执行连接测试close_on_failure保证失败时释放资源_fetch_directories递归列出目录树listdir_attrstat.S_ISDIR判断构建SftpDirectoryInfo层级缓存_fetch_all_files列出各目录含根目录下的文件构建SftpFileInfo文件名、全路径、大小、修改时间、MIME 类型缓存_sort_directories_by_hierarchy按父目录优先的 DFS 顺序排序目录对每个目录生成CreateDirectoryRequest含parent引用以表达嵌套关系对每个文件生成CreateFileRequest关联所属目录、MIME、大小CSV/TSV 附加列 Schema 与可选样本数据管道结束时依据markDeleted*配置执行陈旧实体软删除close()清理缓存并关闭连接。对应的数据模型models.py中SftpFileInfo与SftpDirectoryInfo均为 pydantic 模型字段覆盖名称、全路径、大小、修改时间、MIME 类型等是后续实体构造的基础。八、补充说明本连接器文档对应的示例数据可见 drives 目录含服务、目录、文件、电子表格的样例 JSON可作为理解 Drive 类实体结构的参考仓库中的 SFTP 连接器遵循 OpenMetadata 通用连接器规范其服务类型在 driveService.json 中统一注册配置私钥、口令等敏感信息时建议通过环境变量或 OpenMetadata 的密钥管理机制注入避免明文写入管道配置文件。【免费下载链接】OpenMetadataThe Open Context Layer for Data and AI , OpenMetadata is the open platform for building trusted data context and business semantics for humans, AI assistants, and agents.项目地址: https://gitcode.com/GitHub_Trending/op/OpenMetadata创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
