Restructure Kafka 连接器输出文件

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

Kafka Connector流式传输的数据将转换为RADAR-base导向的输出目录，按项目、用户和收集日期组织。它支持由RADAR HDFS 接收器连接器写入的数据，这些数据仅基于主题名称流式传输到文件中。此工具将该输出转换为本地目录结构：projectId/userId/topic/date_hour.csv。日期和小时从每条记录的time字段提取，并以UTC时间格式化。该工具包含在RADAR-Docker仓库的dcompose/radar-cp-hadoop-stack/bin/hdfs-restructure脚本中。

升级说明

升级到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: [***]  # 本地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镜像radarbase/radar-output-restructure提供。镜像的入口点是当前应用程序。因此，在所有使用命令中，将radar-output-restructure替换为例如：

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。启用前请参阅issue #16。也可在配置文件中按主题启用或禁用去重。如果需要使用部分字段进行去重（例如，仅sourceId和time定义唯一记录，且仅保留具有重复值的最后一条记录），则指定topics: <topicName>: 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使用[***]
    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>)方法。