R2 接收端

R2 sink 将 pipeline 处理后的数据作为原始文件写入 R2 对象存储。目前支持写入 JSON 和 Parquet 格式。

要创建 R2 sink，请运行 pipelines sinks create 命令并指定 sink 类型和目标存储桶：

npx wrangler pipelines sinks create my-sink \
  --type r2 \
  --bucket my-bucket

R2 sink 支持两种输出格式：

将数据写入换行分隔的 JSON 文件：

--format json

将数据写入 Parquet 文件以获得更好的查询性能和压缩：

--format parquet --compression zstd

Parquet 压缩选项：

• zstd（默认）- 最佳压缩比
• snappy - 最快的压缩
• gzip - 良好的压缩，广泛支持
• lz4 - 快速压缩，压缩比合理
• uncompressed - 不压缩

行组大小： 行组 ↗ 是 Parquet 文件中存储在一起的一组行，影响内存使用和查询性能。以 MB 为单位配置目标行组大小：

--target-row-group-size 256

文件以 UUID 名称写入分区目录结构中。例如，路径为 analytics 且使用默认分区时：

analytics/year=2025/month=09/day=18/002507a5-d449-48e8-a484-b1bea916102f.parquet

设置存储桶中写入文件的基础目录：

--path analytics/events

R2 sink 使用可配置模式按时间自动分区文件。默认模式为 year=%Y/month=%m/day=%d（Hive 风格分区）。

--partitioning "year=%Y/month=%m/day=%d/hour=%H"

有关可用的格式说明符，请参阅 strftime 文档 ↗。

控制何时将文件写入 R2。根据需求配置：

• 较低值：更频繁写入，文件更小，延迟更低
• 较高值：写入频率更低，文件更大，查询性能更好

设置文件写入频率（默认：300 秒，最小值：10 秒）：

--roll-interval 10  # Write files every 10 seconds

设置创建新文件前的最大文件大小（MB）：

--roll-size 100  # Create new file after 100MB

R2 sink 需要具有对象读写权限的 API 凭证（Access Key ID 和 Secret Access Key）才能向存储桶写入数据。

npx wrangler pipelines sinks create my-sink \
  --type r2 \
  --bucket my-bucket \
  --access-key-id YOUR_ACCESS_KEY_ID \
  --secret-access-key YOUR_SECRET_ACCESS_KEY