在 AWS Glue Python Shell Job 里面使用 PySpark 的方式操作 FTP 和 S3。代码示例如下:
import pysftp
from io import BytesIO
import boto3
# FTP 连接信息
ftp_host = 'ftp.example.com'
ftp_username = 'username'
ftp_password = 'password'
ftp_file_path = '/path/to/ftp/file.csv'
# S3 存储桶和目标文件路径
s3_bucket_name = 'my-s3-bucket'
s3_file_path = 'path/to/s3/file.csv'
# 通过 PySpark 协议读取 FTP 文件内容并上传到 S3
def upload_to_s3(ftp_file_path, s3_file_path, s3_bucket_name):
with pysftp.Connection(host=ftp_host, username=ftp_username, password=ftp_password) as sftp:
with BytesIO() as bio:
sftp.getfo(ftp_file_path, bio.write)
bio.seek(0)
session = boto3.Session()
s3 = session.resource('s3')
s3.Bucket(s3_bucket_name).Object(s3_file_path).put(Body=bio.read())
upload_to_s3(ftp_file_path, s3_file_path, s3_bucket_name)
在 PySpark Console 里面运行该代码,即可将 FTP 文件下载并上传到 S3 中的指定路径。