Apache Beam/Dataflow用FTPS导数据到BigQuery遇pickle错误求解决方案

首先明确：Apache Beam完全支持从FTPS服务器导入数据，你遇到的TypeError: can't pickle SSLContext objects错误是因为Dataflow的分布式执行特性导致的——Beam需要将代码和对象序列化（pickle）后发送到worker节点，但FTP_TLS实例内部包含的SSLContext对象无法被序列化。直接在主进程创建FTP_TLS实例再传递给worker的做法是行不通的，下面是具体的解决思路和实现方案：

核心修复思路 #

不要在Pipeline定义的主进程中创建FTP_TLS连接，而是将连接逻辑放到worker节点本地执行的代码中——比如在DoFn的setup或process方法里初始化连接，这样每个worker自己创建和管理FTPS连接，完全避开序列化问题。

具体实现步骤 #

1. 使用DoFn的setup方法初始化FTPS连接 #

setup方法会在每个worker实例启动时执行一次，非常适合初始化长连接（比如FTPS），比每次处理数据都创建连接更高效。下面是完整的示例代码：

import apache_beam as beam
from ftplib import FTP_TLS
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions, StandardOptions

class FTPSReaderDoFn(beam.DoFn):
    def __init__(self, ftp_host, ftp_user, ftp_password):
        # 只传递配置参数（可序列化的基本类型），不传递FTPS实例
        self.ftp_host = ftp_host
        self.ftp_user = ftp_user
        self.ftp_password = ftp_password
        self.ftps = None

    def setup(self):
        # 在worker节点本地初始化FTPS连接
        self.ftps = FTP_TLS(self.ftp_host)
        self.ftps.login(self.ftp_user, self.ftp_password)
        # 根据FTPS服务器配置设置被动模式（大多数公共服务器需要）
        self.ftps.set_pasv(True)
        # 切换到加密的数据传输通道
        self.ftps.prot_p()

    def process(self, file_path):
        # 读取指定FTPS文件并转换为BigQuery兼容的格式
        try:
            file_content = []
            self.ftps.retrlines(f'RETR {file_path}', file_content.append)
            # 这里可以根据你的数据格式解析每行内容，比如分割成字段
            for line in file_content:
                yield {'raw_data': line}
        except Exception as e:
            # 处理连接断开等异常，重新初始化连接并重试
            print(f"Failed to read {file_path}: {str(e)}, reconnecting...")
            self.setup()
            # 重试读取
            file_content = []
            self.ftps.retrlines(f'RETR {file_path}', file_content.append)
            for line in file_content:
                yield {'raw_data': line}

def run_pipeline():
    # 配置Dataflow Pipeline选项
    pipeline_options = PipelineOptions()
    gcp_options = pipeline_options.view_as(GoogleCloudOptions)
    gcp_options.project = "your-gcp-project-id"
    gcp_options.job_name = "ftps-to-bq-ingestion"
    gcp_options.staging_location = "gs://your-bucket/staging"
    gcp_options.temp_location = "gs://your-bucket/temp"
    pipeline_options.view_as(StandardOptions).runner = "DataflowRunner"

    # 要读取的FTPS文件路径列表（可从配置或FTPS目录列表生成）
    target_files = ["/data/2024/05/file1.csv", "/data/2024/05/file2.csv"]

    with beam.Pipeline(options=pipeline_options) as p:
        (p
         | "Generate File Paths" >> beam.Create(target_files)
         | "Read from FTPS" >> beam.ParDo(FTPSReaderDoFn(
             ftp_host="ftp.xxxxx.xxx",
             ftp_user="your-ftp-username",
             ftp_password="your-ftp-password"
         ))
         | "Write to BigQuery" >> beam.io.WriteToBigQuery(
             table="your-project:your-dataset.your-table",
             schema="raw_data:STRING",
             write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
             create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED
         ))

if __name__ == "__main__":
    run_pipeline()

2. 优化setup.py配置 #

ftplib是Python标准库，不需要在setup.py中声明依赖，你只需要确保包含Apache Beam的GCP依赖即可：

from setuptools import setup, find_packages

setup(
    name="ftps-to-bq-pipeline",
    version="0.1.0",
    packages=find_packages(),
    install_requires=[
        "apache-beam[gcp]>=2.40.0",
        # 如果你用了其他第三方库，在这里添加
    ],
)

3. 额外注意事项 #

• 敏感信息管理：不要硬编码FTP密码，建议使用GCP Secret Manager存储，通过apache_beam.io.gcp.secretsmanager.SecretsManagerAccessor在worker中安全读取。
• 网络连通性：确保Dataflow worker所在的VPC允许出站访问FTPS服务器的端口（默认是990，被动模式可能需要开放额外端口范围），需要配置对应的防火墙规则。
• 连接复用：setup方法创建的连接会在worker的生命周期内复用，避免频繁创建连接带来的开销；如果连接超时，通过process方法中的异常处理重新初始化即可。

内容的提问来源于stack exchange，提问作者nikhil sahai