senzing/stream-producer

stream-producer

概要

向队列填充记录，供https://github.com/Senzing/stream-loader%E6%B6%88%E8%B4%B9%E3%80%82

概述

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

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

console
$ ./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}
                          ...

队列消息工具。更多信息，请参见 https://github.com/Senzing/stream-producer

位置参数:
  {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，我们致力于以“https://github.com/Senzing/knowledge-base/blob/main/WHATIS/dont-make-me-think.md%E2%80%9D%E7%9A%84%E9%A3%8E%E6%A0%BC%E5%88%9B%E5%BB%BAGitHub%E6%96%87%E6%A1%A3%E3%80%82%E5%A4%A7%E5%A4%9A%E6%95%B0%E6%83%85%E5%86%B5%E4%B8%8B%EF%BC%8C%E6%8C%87%E4%BB%A4%E6%98%AF%E5%A4%8D%E5%88%B6%E7%B2%98%E8%B4%B4%E5%8D%B3%E5%8F%AF%E6%89%A7%E8%A1%8C%E7%9A%84%E3%80%82%E5%BD%93%E9%9C%80%E8%A6%81%E9%A2%9D%E5%A4%96%E6%80%9D%E8%80%83%E6%97%B6%EF%BC%8C%E4%BC%9A%E6%A0%87%E8%AE%B0%E4%B8%BA%E2%80%9C%E6%80%9D%E8%80%83%E2%80%9D%E5%9B%BE%E6%A0%87 :thinking:。当需要自定义时，会标记为“铅笔”图标 :pencil2:。如果指令不清晰，请通过创建新的https://github.com/Senzing/template-python/issues/new?template=documentation_request.md%E5%91%8A%E8%AF%89%E6%88%91%E4%BB%AC%E9%9C%80%E8%A6%81%E6%94%B9%E8%BF%9B%E7%9A%84%E5%9C%B0%E6%96%B9%E3%80%82%E7%8E%B0%E5%9C%A8%E5%BC%80%E5%A7%8B%E6%BC%94%E7%A4%BA...

图例

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

预期

• 空间：本仓库和演示需要6 GB可用磁盘空间。
• 时间：根据CPU和网络速度，预计需要40分钟来启动并运行演示。
• 背景知识：本仓库假设您具备以下知识：
  • https://github.com/Senzing/knowledge-base/blob/main/WHATIS/docker.md

使用Docker演示

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

   console
   docker run \
     --rm \
     docker.xuanyuan.run/senzing/stream-producer --help
   

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

使用docker-compose演示

1. 部署Stream Loader所需的https://github.com/Senzing/knowledge-base/blob/main/HOWTO/deploy-rabbitmq-postgresql-backing-services.md#using-docker-compose%E3%80%82

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

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

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

   console
   curl -X GET \
     --output ${SENZING_VOLUME}/docker-compose.yaml \
     https://raw.githubusercontent.com/Senzing/stream-producer/main/docker-compose.yaml
   

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

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

使用命令行界面演示

CLI先决条件

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

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

   console
   pip3 install -r https://raw.githubusercontent.com/Senzing/stream-producer/main/requirements.txt
   

   1. 参见requirements.txt获取列表。
      1. https://github.com/Senzing/knowledge-base/blob/main/HOWTO/install-python-dependencies.md

下载

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

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

      console
      export SENZING_DOWNLOAD_FILE=~/stream-producer.py
      

   2. 下载文件。示例：

      console
      curl -X GET \
        --output ${SENZING_DOWNLOAD_FILE} \
        https://raw.githubusercontent.com/Senzing/stream-producer/main/stream-producer.py
      

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

      console
      chmod +x ${SENZING_DOWNLOAD_FILE}
      

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

运行命令

1. 运行命令。示例：

   console
   ${SENZING_DOWNLOAD_FILE} --help
   

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

配置

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

• https://github.com/Senzing/knowledge-base/blob/main/lists/environment-variables.md#senzing_network

AWS配置

stream-producer.py使用https://aws.amazon.com/sdk-for-python/%E8%AE%BF%E9%97%AEAWS%E6%9C%8D%E5%8A%A1%E3%80%82%E5%8F%AF%E9%80%9A%E8%BF%87%E7%8E%AF%E5%A2%83%E5%8F%98%E9%87%8F%E6%88%96%60~/.aws/config%60%E6%96%87%E4%BB%B6%E9%85%8D%E7%BD%AE%E6%AD%A4%E5%BA%93%E3%80%82

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

• https://github.com/Senzing/knowledge-base/blob/main/lists/environment-variables.md#aws_access_key_id
• https://github.com/Senzing/knowledge-base/blob/main/lists/environment-variables.md#aws_secret_access_key
• https://github.com/Senzing/knowledge-base/blob/main/lists/environment-variables.md#aws_default_region

参考资料

1. Boto3 https://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html
2. Boto3 https://boto3.amazonaws.com/v1/documentation/api/latest/guide/credentials.html
3. 开发
4. 错误
5. 示例
6. 相关工件：
   1. https://hub.docker.com/r/senzing/stream-producer