radarbase/radar-output-restructure

Restructure Kafka 连接器输出文件

![构建状态]([***]

Kafka Connector流式传输的数据将转换为RADAR-base导向的输出目录，按项目、用户和收集日期组织。它支持由https://github.com/RADAR-base/RADAR-HDFS-Sink-Connector%E5%86%99%E5%85%A5%E7%9A%84%E6%95%B0%E6%8D%AE%EF%BC%8C%E8%BF%99%E4%BA%9B%E6%95%B0%E6%8D%AE%E4%BB%85%E5%9F%BA%E4%BA%8E%E4%B8%BB%E9%A2%98%E5%90%8D%E7%A7%B0%E6%B5%81%E5%BC%8F%E4%BC%A0%E8%BE%93%E5%88%B0%E6%96%87%E4%BB%B6%E4%B8%AD%E3%80%82%E6%AD%A4%E5%B7%A5%E5%85%B7%E5%B0%86%E8%AF%A5%E8%BE%93%E5%87%BA%E8%BD%AC%E6%8D%A2%E4%B8%BA%E6%9C%AC%E5%9C%B0%E7%9B%AE%E5%BD%95%E7%BB%93%E6%9E%84%EF%BC%9A%60projectId/userId/topic/date_hour.csv%60%E3%80%82%E6%97%A5%E6%9C%9F%E5%92%8C%E5%B0%8F%E6%97%B6%E4%BB%8E%E6%AF%8F%E6%9D%A1%E8%AE%B0%E5%BD%95%E7%9A%84%60time%60%E5%AD%97%E6%AE%B5%E6%8F%90%E5%8F%96%EF%BC%8C%E5%B9%B6%E4%BB%A5UTC%E6%97%B6%E9%97%B4%E6%A0%BC%E5%BC%8F%E5%8C%96%E3%80%82%E8%AF%A5%E5%B7%A5%E5%85%B7%E5%8C%85%E5%90%AB%E5%9C%A8https://github.com/RADAR-base/RADAR-Docker%E4%BB%93%E5%BA%93%E7%9A%84%60dcompose/radar-cp-hadoop-stack/bin/hdfs-restructure%60%E8%84%9A%E6%9C%AC%E4%B8%AD%E3%80%82

升级说明

升级到1.2.0版本时，请遵循以下说明：

• 使用本地目标存储时，确保：
  1. 存储可由用户101写入，或使用docker命令行标志--user更改运行时用户为可写入目标存储的用户；
  2. 本地存储属性userId和groupId设置为可写入目标存储的值。

从0.6.0版本升级到1.0.0或更高版本时，请遵循以下说明：

• 该工具现在依赖Redis进行锁定和偏移量管理。请安装Redis或使用docker-compose.yml文件启动它。
• 编写restructure.yml配置文件以匹配0.6.0版本使用的设置：

  • HDFS设置已移至source。在nameNodes属性中指定所有名称节点。name属性不再使用。

    yaml
    source:
      type: hdfs
      hdfs:
        nameNodes: [hdfs-namenode]
    

  • 添加redis块：

    yaml
    redis:
      uri: redis://localhost:6379
    

  • 当radar-output处理主题时，偏移量会计将自动从基于文件的存储迁移到Redis条目。请在偏移量目录为空之前不要删除它。

  • 存储设置已移至target块。使用本地输出目录：

    yaml
    target:
      type: local
      local:
        # 写入数据的用户ID。仅当显式将运行时用户设置为root时才有效。
        userId: 123
        # 写入数据的组ID。仅当显式将运行时用户设置为root时才有效。
        groupId: 123
    

    使用S3StorageDriver时，使用以下配置：

    yaml
    target:
      type: s3
      s3:
        endpoint: https://my-region.s3.aws.amazon.com  # 本地minio使用http://localhost:9000
        accessToken: ABA...
        secretKey: CSD...
        bucket: myBucketName
    

从0.5.x或更早版本升级到0.6.0版本时，请遵循以下说明：

• 编写restructure.yml配置文件以匹配0.5.x版本使用的命令行设置。
• 如有需要，将offsets.csv的所有条目移动到offsets/<topic>.csv中的每个主题文件。首先进入输出目录，然后运行bin/migrate-offsets-to-0.6.0.sh脚本。

Docker使用

该工具以Docker镜像https://hub.docker.com/r/radarbase/radar-output-restructure%E6%8F%90%E4%BE%9B%E3%80%82%E9%95%9C%E5%83%8F%E7%9A%84%E5%85%A5%E5%8F%A3%E7%82%B9%E6%98%AF%E5%BD%93%E5%89%8D%E5%BA%94%E7%94%A8%E7%A8%8B%E5%BA%8F%E3%80%82%E5%9B%A0%E6%AD%A4%EF%BC%8C%E5%9C%A8%E6%89%80%E6%9C%89%E4%BD%BF%E7%94%A8%E5%91%BD%E4%BB%A4%E4%B8%AD%EF%BC%8C%E5%B0%86%60radar-output-restructure%60%E6%9B%BF%E6%8D%A2%E4%B8%BA%E4%BE%8B%E5%A6%82%EF%BC%9A

shell
docker run --rm -t --network hadoop -v "$PWD/output:/output" radarbase/radar-output-restructure:1.2.0 -n hdfs-namenode -o /output /myTopic

命令行使用

安装应用程序后，可按如下方式使用：

shell
radar-output-restructure --nameservice <hdfs_node> --output-directory <output_folder> <input_path_1> [<input_path_2> ...]

或使用简短形式：

shell
radar-output-restructure -n <hdfs_node> -o <output_folder> <input_path_1> [<input_path_2> ...]

要显示使用方法和所有可用选项，可使用帮助选项：

shell
radar-output-restructure --help

注意，上述输出中前面带*的选项是运行应用程序所需的必填项。此外，可以有多个输入路径从中读取文件。例如 - /topicAndroidNew/topic1 /topicAndroidNew/topic2 ...。至少提供一个输入路径。

每个参数以及更多选项都可以在配置文件中提供。配置文件的默认名称为restructure.yml。请参考当前目录中的restructure.yml了解所有可用选项。可使用-F标志指定替代文件。

文件格式

默认情况下，输出数据为CSV格式。如果需要JSON格式，可使用以下命令：

shell
radar-output-restructure --format json --nameservice <hdfs_node> --output-directory <output_folder>  <input_path_1> [<input_path_2> ...]

默认情况下，写入后不 deduplicate 文件记录。要启用此行为，指定选项--deduplicate或-d。由于Biovotion数据的问题，默认情况下此选项为false。启用前请参阅https://github.com/RADAR-base/Restructure-HDFS-topic/issues/16%E3%80%82%E4%B9%9F%E5%8F%AF%E5%9C%A8%E9%85%8D%E7%BD%AE%E6%96%87%E4%BB%B6%E4%B8%AD%E6%8C%89%E4%B8%BB%E9%A2%98%E5%90%AF%E7%94%A8%E6%88%96%E7%A6%81%E7%94%A8%E5%8E%BB%E9%87%8D%E3%80%82%E5%A6%82%E6%9E%9C%E9%9C%80%E8%A6%81%E4%BD%BF%E7%94%A8%E9%83%A8%E5%88%86%E5%AD%97%E6%AE%B5%E8%BF%9B%E8%A1%8C%E5%8E%BB%E9%87%8D%EF%BC%88%E4%BE%8B%E5%A6%82%EF%BC%8C%E4%BB%85%60sourceId%60%E5%92%8C%60time%60%E5%AE%9A%E4%B9%89%E5%94%AF%E4%B8%80%E8%AE%B0%E5%BD%95%EF%BC%8C%E4%B8%94%E4%BB%85%E4%BF%9D%E7%95%99%E5%85%B7%E6%9C%89%E9%87%8D%E5%A4%8D%E5%80%BC%E7%9A%84%E6%9C%80%E5%90%8E%E4%B8%80%E6%9D%A1%E8%AE%B0%E5%BD%95%EF%BC%89%EF%BC%8C%E5%88%99%E6%8C%87%E5%AE%9A%60topics: : deduplication: distinctFields: [key.sourceId, value.time]`。

压缩

另一个选项是以压缩形式输出数据。所有文件将获得gz后缀，可使用GZIP解码器解压缩。注意，对于少量记录，压缩可能会增加文件大小。也支持Zip压缩。

radar-output-restructure --compression gzip  --nameservice <hdfs_node> --output-directory <output_folder> <input_path_1> [<input_path_2> ...]

Redis

该工具假设运行有Redis服务。有关配置选项，请参见示例restructure.yml。

源和目标

source和target属性包含资源描述。源可以有两种类型：hdfs和s3：

yaml
source:
  type: s3  # hdfs 或 s3
  s3:
    endpoint: http://localhost:9000  # 也可使用AWS S3端点
    bucket: radar
    accessToken: minioadmin
    secretKey: minioadmin
  # 仅当源类型为hdfs时需要
  hdfs:
    nameNodes: [hdfs-namenode-1, hdfs-namenode-2]

目标类似，但不支持HDFS，支持本地文件系统（local）或s3。

yaml
target:
  type: s3  # s3 或 local
  s3:
    endpoint: http://localhost:9000  # 本地minio使用http://localhost:9000，AWS S3使用https://my-region.s3.aws.amazon.com
    bucket: out
    accessToken: minioadmin
    secretKey: minioadmin
  # 仅当目标类型为local时需要
  local:
    userId: 1000  # 写入数据的用户ID，默认-1使用当前用户
    groupId: 100  # 写入数据的组ID，默认-1使用当前用户

也可通过环境变量提供密钥：

将上述变量中的SOURCE替换为TARGET可配置目标存储。

清理器

清理器进程可自动删除源文件。它检查文件是否已提取且超过配置的老化时间。默认情况下未启用此功能。可在cleaner配置部分中配置：

yaml
cleaner:
  # 启用清理旧源文件
  enable: true
  # 清理数据的间隔（秒）
  interval: 1260  # 21分钟
  # 源文件被视为旧文件的天数
  age: 7

也可使用--cleaner命令行标志启用清理器。要将清理器作为与输出重组分离的独立进程运行，启动配置属性worker: enable: false或命令行参数--no-restructure的进程。

服务模式

要将输出生成器作为定期轮询HDFS目录的服务运行，添加--service标志，并可选择使用--interval标志调整轮询间隔，或使用相应的配置文件参数。

本地构建

该工具至少需要Java JDK 8。使用以下命令构建发行版：

shell
./gradlew build

并将包安装到/usr/local，例如：

shell
sudo mkdir -p /usr/local
sudo tar -xzf build/distributions/radar-output-restructure-1.2.0.tar.gz -C /usr/local --strip-components=1

现在radar-output-restructure命令应该可用。

扩展连接器

要实现替代存储路径、存储驱动程序或存储格式，将自定义JAR放在$APP_DIR/lib/radar-output-plugins中。要加载它们，使用以下选项：

相应的<type>: properties: {}配置参数可用于提供工厂的自定义配置。此配置将传递给Plugin#init(Map<String, String>)方法。