尧图网站建设 
如何把 Airbyte 数据源接入 Pathway 构建实时 ETL 管道【免费下载链接】pathwayPython ETL framework for stream processing, real-time analytics, LLM pipelines, and RAG.项目地址: https://gitcode.com/GitHub_Trending/pa/pathwayPathway Live Data Framework 自带 Airbyte 输入连接器可以把 Airbyte 的源连接器当作 Pathway 的输入端Airbyte 负责 extractPathway 用 Python 代码完成 transform 和 load。本文以文档中的示例任务为准——用 Airbyte 的 GitHub 连接器实时读取某个仓库的 commits 流在管道中剔除 payload 里的邮箱地址去 PII再把结果写入本地文件。这条路径同样适用于其他 Airbyte 免费源连接器。前提条件均来自项目文档Pathway 支持 MacOS 和 Linux不支持 Windowsairbyte-serverless工具需要本机已安装 Docker使用 GitHub 连接器需要一个 PATPersonal Access Token需要public_repo访问公共仓库范围可在 GitHub 的 Tokens 页面获取。安装依赖先安装 Pathway 及其 Airbyte 可选包再安装airbyte-serverless工具用于生成和运行 Airbyte 源配置pip install -U pathway pip install pathway[airbyte] pip install airbyte-serverlesspathway[airbyte]是安装文档中列出的 Airbyte Connector 支持包airbyte-serverless需要 Docker 环境。配置 Airbyte 数据源文档给出两种配置方式选其一即可。方式一使用预配置模板在/connections/github.yaml路径下创建 YAML 文件填入以下模板并把personal_access_token替换为你自己的 GitHub PATsource: docker_image: airbyte/source-github # Here the airbyte connector type is specified config: credentials: option_title: PAT Credentials # The second authentication option youve uncommented personal_access_token: YOUR PERSONAL ACCESS TOKEN HERE # Taken from https://github.com/settings/tokens repositories: - pathwaycom/pathway # Pathway repository api_url: https://api.github.com/ streams: commits模板中的YOUR PERSONAL ACCESS TOKEN HERE是占位符必须替换为真实 PATrepositories换成你要读取的仓库streams是 Airbyte 流名GitHub 连接器的提交数据在commits流中。方式二手动生成配置模板也可以用命令行生成 GitHub 源的配置模板abs create github --source airbyte/source-github生成的文件是./connections/github.yaml。此时模板尚未可用文档明确提醒如果不补全配置并删除无用字段airbyte会直接报错。需要做的修改选择 PAT 认证删除未注释的option_title、access_token、client_id、client_secret字段然后取消 Another valid structure for credentials 一节的注释填入 PAT在repositories字段设置仓库名例如pathwaycom/pathway删除未使用的可选字段。在 Pathway 中提取数据配置就绪后管道代码的提取部分如下import pathway as pw commits_table pw.io.airbyte.read( ./connections/github.yaml, streams[commits], )pw.io.airbyte.read只需要两个参数配置文件路径和要读取的 Airbyte 流名列表。返回的表包含一个data列类型是pw.Json内容为 Airbyte 格式的 JSON payload。两个影响运行行为的关键参数mode默认streaming会持续轮询源端新 commits 产生后不断追加到表里设为static则只读取当前已存在的数据并在一次 commit 后终止不再等待新数据。refresh_interval控制 streaming 模式下轮询新数据的频率接受秒数或datetime.timedelta默认 60 秒。enforce_method不指定时Pathway 会先在 PyPI 上查找该连接器并在独立虚拟环境中安装运行找不到才回退到 Docker 镜像设为pypi或docker可强制指定。文档建议在部署环境中显式使用enforce_method因为 PyPI 服务不可用时自动探测可能失败。如果只需要一次性加载历史 commits用 static 模式即可commits_table pw.io.airbyte.read( ./connections/github.yaml, streams[commits], modestatic, )转换数据流本例的转换目标是剔除 payload 中的邮箱。文档给出的做法是把pw.Json解析成 Python 对象深度优先遍历 JSON凡是非空白字符组中含有的整组剔除从而以高召回率清掉所有邮箱import json def remove_emails_from_data(payload): if isinstance(payload, str): # The string case is obvious: its getting split and then merged back after # the email-like substrings are removed return .join([item for item in payload.split( ) if not in item]) if isinstance(payload, list): # If the payload is a list, one needs to remove emails from each of its # elements and then return the result of the processing result [] for item in payload: result.append(remove_emails_from_data(item)) return result if isinstance(payload, dict): # If the payload is a dict, one needs to remove emails from its keys and # values and then return the clean dict result {} for key, value in payload.items(): # There are no e-mails in the keys of the returned dict # So, we only need to remove them from values value remove_emails_from_data(value) result[key] value return result # If the payload is neither str nor list or dict, its a primitive type: # namely, a boolean, a float, or an int. It can also be just null. # # But in any case, there is no data to remove from such an element. return payload def remove_emails(raw_commit_data: pw.Json) - pw.Json: # First, parse pw.Json type into a Python dict data json.loads(raw_commit_data.as_str()) # Next, just apply the recursive method to delete e-mails return remove_emails_from_data(data) commits_table commits_table.select(datapw.apply(remove_emails, pw.this.data))转换完全在 Python 中完成通过pw.apply把函数应用到表的data列。执行完后commits_table中的data列即为去除邮箱后的 JSON。加载与运行输出端可以使用 Pathway 的任意输出连接器文档列出的选项包括 Kafka topic、Logstash endpoint、Postgres 表、Python 回调pw.io.subscribe。最简单的验证路径是写入本地 jsonlines 文件pw.io.jsonlines.write(commits_table, commits.jsonlines) pw.run()pw.run()启动管道。按mode的不同预期行为是static 模式读取当前已存在的全部 commits管道处理完并终止结果落盘到commits.jsonlinesstreaming 模式默认管道持续运行每refresh_interval秒轮询一次源端新产生的 commits 会被追加到表里并写入commits.jsonlines。验证方式就是检查commits.jsonlinesstatic 模式下程序正常退出且文件中有对应记录streaming 模式下在源仓库新提交一个 commit 后文件中应出现对应的新记录。排查与限制以下是文档中明确给出的、与这条接入路径直接相关的约束流的同步模式必须一致同一次pw.io.airbyte.read中的所有流必须是相同的sync_mode目前支持incremental和full_refresh混用会抛出ValueErrorAirbyte license 复用不受支持pw.io.airbyte.read的文档说明当前不支持复用 Airbyte license避免 Docker-in-Docker文档不推荐在容器内部再用容器跑 Airbyte 连接器DinD。如果部署环境本身是 Docker 容器建议改用enforce_methodpypi连接器作为 Python 包装在虚拟环境中运行绕开 Docker或改用execution_typeremote把提取部分放到 Google Cloud Run 上执行需要提供service_user_credentials_file指向的服务账号凭据文件且 Cloud Run 按 vCPU 时间和内存时间计费不建议把refresh_interval设得太小Windows 不受支持Pathway 目前仅在 MacOS 和 Linux 上可用。完整的步骤与代码可对照仓库内文档ETL 模板Airbyte Pathway、Airbyte 连接器说明和安装文档。【免费下载链接】pathwayPython ETL framework for stream processing, real-time analytics, LLM pipelines, and RAG.项目地址: https://gitcode.com/GitHub_Trending/pa/pathway创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 如何把 Airbyte 数据源接入 Pathway 构建实时 ETL 管道？
如何把 Airbyte 数据源接入 Pathway 构建实时 ETL 管道【免费下载链接】pathwayPython ETL framework for stream processing, real-time analytics, LLM pipelines, and RAG.项目地址: https://gitcode.com/GitHub_Trending/pa/pathwayPathway Live Data Framework 自带 Airbyte 输入连接器可以把 Airbyte 的源连接器当作 Pathway 的输入端Airbyte 负责 extractPathway 用 Python 代码完成 transform 和 load。本文以文档中的示例任务为准——用 Airbyte 的 GitHub 连接器实时读取某个仓库的 commits 流在管道中剔除 payload 里的邮箱地址去 PII再把结果写入本地文件。这条路径同样适用于其他 Airbyte 免费源连接器。前提条件均来自项目文档Pathway 支持 MacOS 和 Linux不支持 Windowsairbyte-serverless工具需要本机已安装 Docker使用 GitHub 连接器需要一个 PATPersonal Access Token需要public_repo访问公共仓库范围可在 GitHub 的 Tokens 页面获取。安装依赖先安装 Pathway 及其 Airbyte 可选包再安装airbyte-serverless工具用于生成和运行 Airbyte 源配置pip install -U pathway pip install pathway[airbyte] pip install airbyte-serverlesspathway[airbyte]是安装文档中列出的 Airbyte Connector 支持包airbyte-serverless需要 Docker 环境。配置 Airbyte 数据源文档给出两种配置方式选其一即可。方式一使用预配置模板在/connections/github.yaml路径下创建 YAML 文件填入以下模板并把personal_access_token替换为你自己的 GitHub PATsource: docker_image: airbyte/source-github # Here the airbyte connector type is specified config: credentials: option_title: PAT Credentials # The second authentication option youve uncommented personal_access_token: YOUR PERSONAL ACCESS TOKEN HERE # Taken from https://github.com/settings/tokens repositories: - pathwaycom/pathway # Pathway repository api_url: https://api.github.com/ streams: commits模板中的YOUR PERSONAL ACCESS TOKEN HERE是占位符必须替换为真实 PATrepositories换成你要读取的仓库streams是 Airbyte 流名GitHub 连接器的提交数据在commits流中。方式二手动生成配置模板也可以用命令行生成 GitHub 源的配置模板abs create github --source airbyte/source-github生成的文件是./connections/github.yaml。此时模板尚未可用文档明确提醒如果不补全配置并删除无用字段airbyte会直接报错。需要做的修改选择 PAT 认证删除未注释的option_title、access_token、client_id、client_secret字段然后取消 Another valid structure for credentials 一节的注释填入 PAT在repositories字段设置仓库名例如pathwaycom/pathway删除未使用的可选字段。在 Pathway 中提取数据配置就绪后管道代码的提取部分如下import pathway as pw commits_table pw.io.airbyte.read( ./connections/github.yaml, streams[commits], )pw.io.airbyte.read只需要两个参数配置文件路径和要读取的 Airbyte 流名列表。返回的表包含一个data列类型是pw.Json内容为 Airbyte 格式的 JSON payload。两个影响运行行为的关键参数mode默认streaming会持续轮询源端新 commits 产生后不断追加到表里设为static则只读取当前已存在的数据并在一次 commit 后终止不再等待新数据。refresh_interval控制 streaming 模式下轮询新数据的频率接受秒数或datetime.timedelta默认 60 秒。enforce_method不指定时Pathway 会先在 PyPI 上查找该连接器并在独立虚拟环境中安装运行找不到才回退到 Docker 镜像设为pypi或docker可强制指定。文档建议在部署环境中显式使用enforce_method因为 PyPI 服务不可用时自动探测可能失败。如果只需要一次性加载历史 commits用 static 模式即可commits_table pw.io.airbyte.read( ./connections/github.yaml, streams[commits], modestatic, )转换数据流本例的转换目标是剔除 payload 中的邮箱。文档给出的做法是把pw.Json解析成 Python 对象深度优先遍历 JSON凡是非空白字符组中含有的整组剔除从而以高召回率清掉所有邮箱import json def remove_emails_from_data(payload): if isinstance(payload, str): # The string case is obvious: its getting split and then merged back after # the email-like substrings are removed return .join([item for item in payload.split( ) if not in item]) if isinstance(payload, list): # If the payload is a list, one needs to remove emails from each of its # elements and then return the result of the processing result [] for item in payload: result.append(remove_emails_from_data(item)) return result if isinstance(payload, dict): # If the payload is a dict, one needs to remove emails from its keys and # values and then return the clean dict result {} for key, value in payload.items(): # There are no e-mails in the keys of the returned dict # So, we only need to remove them from values value remove_emails_from_data(value) result[key] value return result # If the payload is neither str nor list or dict, its a primitive type: # namely, a boolean, a float, or an int. It can also be just null. # # But in any case, there is no data to remove from such an element. return payload def remove_emails(raw_commit_data: pw.Json) - pw.Json: # First, parse pw.Json type into a Python dict data json.loads(raw_commit_data.as_str()) # Next, just apply the recursive method to delete e-mails return remove_emails_from_data(data) commits_table commits_table.select(datapw.apply(remove_emails, pw.this.data))转换完全在 Python 中完成通过pw.apply把函数应用到表的data列。执行完后commits_table中的data列即为去除邮箱后的 JSON。加载与运行输出端可以使用 Pathway 的任意输出连接器文档列出的选项包括 Kafka topic、Logstash endpoint、Postgres 表、Python 回调pw.io.subscribe。最简单的验证路径是写入本地 jsonlines 文件pw.io.jsonlines.write(commits_table, commits.jsonlines) pw.run()pw.run()启动管道。按mode的不同预期行为是static 模式读取当前已存在的全部 commits管道处理完并终止结果落盘到commits.jsonlinesstreaming 模式默认管道持续运行每refresh_interval秒轮询一次源端新产生的 commits 会被追加到表里并写入commits.jsonlines。验证方式就是检查commits.jsonlinesstatic 模式下程序正常退出且文件中有对应记录streaming 模式下在源仓库新提交一个 commit 后文件中应出现对应的新记录。排查与限制以下是文档中明确给出的、与这条接入路径直接相关的约束流的同步模式必须一致同一次pw.io.airbyte.read中的所有流必须是相同的sync_mode目前支持incremental和full_refresh混用会抛出ValueErrorAirbyte license 复用不受支持pw.io.airbyte.read的文档说明当前不支持复用 Airbyte license避免 Docker-in-Docker文档不推荐在容器内部再用容器跑 Airbyte 连接器DinD。如果部署环境本身是 Docker 容器建议改用enforce_methodpypi连接器作为 Python 包装在虚拟环境中运行绕开 Docker或改用execution_typeremote把提取部分放到 Google Cloud Run 上执行需要提供service_user_credentials_file指向的服务账号凭据文件且 Cloud Run 按 vCPU 时间和内存时间计费不建议把refresh_interval设得太小Windows 不受支持Pathway 目前仅在 MacOS 和 Linux 上可用。完整的步骤与代码可对照仓库内文档ETL 模板Airbyte Pathway、Airbyte 连接器说明和安装文档。【免费下载链接】pathwayPython ETL framework for stream processing, real-time analytics, LLM pipelines, and RAG.项目地址: https://gitcode.com/GitHub_Trending/pa/pathway创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考