stream-producer

概要

向队列填充记录，供stream-loader消费。

概述

stream-producer.py Python脚本可读取不同格式的文件（JSON、CSV、Parquet、Avro）并将其发布到队列（RabbitMQ、Kafka、AWS SQS）。senzing/stream-producer Docker镜像作为封装，用于Docker部署形式（如docker-compose、kubernetes）。

要查看所有子命令，请运行：

$ ./stream-producer.py --help
usage: stream-producer.py [-h]
                          {avro-to-kafka,avro-to-rabbitmq,avro-to-sqs,avro-to-sqs-batch,avro-to-stdout,csv-to-kafka,csv-to-rabbitmq,csv-to-sqs,csv-to-sqs-batch,csv-to-stdout,gzipped-json-to-kafka,gzipped-json-to-rabbitmq,gzipped-json-to-sqs,gzipped-json-to-sqs-batch,gzipped-json-to-stdout,json-to-kafka,json-to-rabbitmq,json-to-sqs,json-to-sqs-batch,json-to-stdout,parquet-to-kafka,parquet-to-rabbitmq,parquet-to-sqs,parquet-to-sqs-batch,parquet-to-stdout,websocket-to-kafka,websocket-to-rabbitmq,websocket-to-sqs,websocket-to-sqs-batch,websocket-to-stdout,sleep,version,docker-acceptance-test}
                          ...

队列消息工具。更多信息，请参见 [***]

位置参数:
  {avro-to-kafka,avro-to-rabbitmq,avro-to-sqs,avro-to-sqs-batch,avro-to-stdout,csv-to-kafka,csv-to-rabbitmq,csv-to-sqs,csv-to-sqs-batch,csv-to-stdout,gzipped-json-to-kafka,gzipped-json-to-rabbitmq,gzipped-json-to-sqs,gzipped-json-to-sqs-batch,gzipped-json-to-stdout,json-to-kafka,json-to-rabbitmq,json-to-sqs,json-to-sqs-batch,json-to-stdout,parquet-to-kafka,parquet-to-rabbitmq,parquet-to-sqs,parquet-to-sqs-batch,parquet-to-stdout,websocket-to-kafka,websocket-to-rabbitmq,websocket-to-sqs,websocket-to-sqs-batch,websocket-to-stdout,sleep,version,docker-acceptance-test}
                              子命令（SENZING_SUBCOMMAND）:
    avro-to-kafka             读取Avro文件并发送到Kafka
    avro-to-rabbitmq          读取Avro文件并发送到RabbitMQ
    avro-to-sqs               读取Avro文件并发送到AWS SQS
    avro-to-stdout            读取Avro文件并输出到STDOUT

    csv-to-kafka              读取CSV文件并发送到Kafka
    csv-to-rabbitmq           读取CSV文件并发送到RabbitMQ
    csv-to-sqs                读取CSV文件并发送到SQS
    csv-to-stdout             读取CSV文件并输出到STDOUT

    gzipped-json-to-kafka     读取gzip压缩的JSON文件并发送到Kafka
    gzipped-json-to-rabbitmq  读取gzip压缩的JSON文件并发送到RabbitMQ
    gzipped-json-to-sqs       读取gzip压缩的JSON文件并发送到AWS SQS
    gzipped-json-to-stdout    读取gzip压缩的JSON文件并输出到STDOUT

    json-to-kafka             读取JSON文件并发送到Kafka
    json-to-rabbitmq          读取JSON文件并发送到RabbitMQ
    json-to-sqs               读取JSON文件并发送到AWS SQS
    json-to-stdout            读取JSON文件并输出到STDOUT

    parquet-to-kafka          读取Parquet文件并发送到Kafka
    parquet-to-rabbitmq       读取Parquet文件并发送到RabbitMQ
    parquet-to-sqs            读取Parquet文件并发送到AWS SQS
    parquet-to-stdout         读取Parquet文件并输出到STDOUT

    sleep                     仅休眠，用于Docker测试
    version                   打印程序版本
    docker-acceptance-test    用于Docker验收测试

可选参数:
  -h, --help                  显示此帮助消息并退出

目录

1. 使用Docker演示
2. 使用docker-compose演示
3. 使用命令行界面演示
   1. CLI先决条件
   2. 下载
   3. 运行命令
4. 配置
5. AWS配置
6. 参考资料

前言

在Senzing，我们致力于以“别让我思考”的风格创建GitHub文档。大多数情况下，指令是复制粘贴即可执行的。当需要额外思考时，会标记为“思考”图标 :thinking:。当需要自定义时，会标记为“铅笔”图标 :pencil2:。如果指令不清晰，请通过创建新的文档问题告诉我们需要改进的地方。现在开始演示...

图例

1. :thinking: - “思考者”图标表示可能需要一些额外思考。可能需要做出选择，也可能是可选步骤。
2. :pencil2: - “铅笔”图标表示执行前可能需要修改指令。
3. :warning: - “警告”图标表示有需要注意的复杂操作。

预期

• 空间：本仓库和演示需要6 GB可用磁盘空间。
• 时间：根据CPU和网络速度，预计需要40分钟来启动并运行演示。
• 背景知识：本仓库假设您具备以下知识：
  • Docker

使用Docker演示

1. 运行Docker容器。以下命令将显示帮助信息。示例：

   docker run \
     --rm \
     senzing/stream-producer --help
   

2. 更多使用示例，请参见Docker示例。

使用docker-compose演示

1. 部署Stream Loader所需的支持服务。

2. 指定放置工件的目录。示例：

   export SENZING_VOLUME=~/my-senzing
   mkdir -p ${SENZING_VOLUME}
   

3. 下载docker-compose.yaml文件。示例：

   curl -X GET \
     --output ${SENZING_VOLUME}/docker-compose.yaml \
     [***]
   

4. 启动docker-compose栈。示例：

   docker-compose -f ${SENZING_VOLUME}/docker-compose.yaml up
   

使用命令行界面演示

CLI先决条件

:thinking: 继续之前需要完成以下任务。这些是“一次性任务”，可能已经完成。

1. 安装Python依赖项。示例：

   pip3 install -r [***]
   

   1. 参见requirements.txt获取列表。
      1. 安装提示

下载

1. 获取stream-producer.py的本地副本。示例：

   1. :pencil2: 指定下载文件的位置。示例：

      export SENZING_DOWNLOAD_FILE=~/stream-producer.py
      

   2. 下载文件。示例：

      curl -X GET \
        --output ${SENZING_DOWNLOAD_FILE} \
        [***]
      

   3. 使文件可执行。示例：

      chmod +x ${SENZING_DOWNLOAD_FILE}
      

2. :thinking: 替代方案：可通过克隆仓库中的说明下载整个git仓库。

运行命令

1. 运行命令。示例：

   ${SENZING_DOWNLOAD_FILE} --help
   

2. 更多使用示例，请参见CLI示例。

配置

通过环境变量或命令行参数指定配置值。

• SENZING_NETWORK

AWS配置

stream-producer.py使用AWS SDK for Python (Boto3)访问AWS服务。可通过环境变量或~/.aws/config文件配置此库。

用于配置的环境变量示例：

• AWS_ACCESS_KEY_ID
• AWS_SECRET_ACCESS_KEY
• AWS_DEFAULT_REGION

参考资料

1. Boto3 配置
2. Boto3 凭证
3. 开发
4. 错误
5. 示例
6. 相关工件：
   1. DockerHub