在 S3 中创建组文件，而不是从每个 json 中创建一个 - NIFI

Question

我正在接收 API 调用的结果，进行一些转换并将其存储在 S3 中，现在它为每个 api 调用存储 1 个文件。导致文件很多，流程是：

invokeHTTP->Split.json->JoltTransformJSON（我不需要所有数据）->EvaluateJsonPath->InferAvroScheme（500 个样本）->ConvertJSONToAvro->PutS3Object

json格式为：

"data": {"value1": "test", "value2": "test2"},
"actions": [{"buy": 5, "sell": 6},{"buyAgain": 5, "sellAgain": 6}],
"Reactions": [{"buy": 5, "sell": 6}],
{"otherValue": "1",
"otherValue2": "2"}

有时动作内部有值，在其他情况下“动作”：[] 我使用 JoltTransformJSON 和 remove 参数删除反应，它有很多我不需要的数据

为了加入值，我尝试了 MergeContent，但它丢弃了很多记录，首先我阅读了可能的配置，然后......我开始修改参数只是为了看看它如何改变输出，它总是 DROP 很多记录.

所以现在我在 S3 中每个 json 存储 1 个文件，这是很多文件，您可以在查询数据时感受到它。

如何改进流程以存储更少的文件？ 谢谢！

---- 编辑：添加图片----

当前 MergeContents 配置，不太了解 Attribute strategy 属性。这可以修复架构中的更改吗？ （具有价值的行动或“行动”：[]）

---- 编辑 2 ---- 现在我可以确认这是按预期按状态分组，但删除具有“动作”的 JSON 流：[]，它们具有与该字段已满的某些流相同的状态，有什么想法吗？谢谢！

Answer 1

MergeContent 处理器是这​​里的正确解决方案。将merge format 设置为Avro，流文件的Avro 内容将连接在一起形成一个流文件。您的数据丢失问题与元数据策略属性有关：

对于输入格式支持元数据（例如 Avro）的流文件，此 属性确定应将哪些元数据添加到包中。如果 选择“使用第一个元数据”，元数据键/值来自 将使用要捆绑的第一个 FlowFile。 If 'Keep Only Common Metadata'被选中，只有所有FlowFiles上存在的元数据 在捆绑包中，具有相同的值，将被保留。如果'忽略 选择元数据'，没有元数据传输到传出 捆绑的流文件。如果选择了“不合并不常见的元数据”，则任何 FlowFile 的元数据值与第一个捆绑的不匹配 FlowFile 不会被合并。

架构不等于第一个捆绑流文件的架构的流文件将被删除。我可以想到两种可能的解决方案来防止这种情况：

使用Correlation Attribute Name 合并共享相同架构的 Avro 流文件

您必须确保仅合并具有相同架构的文件。因此，如果您可以在流文件中添加一些属性，例如“type=CAR 或 type=BIKE”，您可以将 Correlation Attribute 设置为“type”。 MergeContent 然后将根据类型制作捆绑包。由于包中文件的架构相同，因此不会删除任何记录。

设置特定架构

将InferAvroSchema 和ConvertJsonToAvro 替换为单个处理器：ConvertRecord。将JsonTreeReader 配置为阅读器并保留默认属性。将AvroRecordSetWriter 配置为编写器并设置以下属性：

在AvroRecordSetWriter配置下面Schema text:

{
  "name": "MyClass",
  "type": "record",
  "namespace": "com.acme.avro",
  "fields": [
    {
      "name": "data",
      "type": {
        "name": "data",
        "type": "record",
        "fields": [
          {
            "name": "value1",
            "type": "string"
          },
          {
            "name": "value2",
            "type": "string"
          }
        ]
      }
    },
    {
      "name": "actions",
      "type": {
        "type": "array",
        "items": {
          "name": "actions_record",
          "type": "record",
          "fields": [
            {
              "name": "buyAgain",
              "type": ["int", "null"]
            },
            {
              "name": "sellAgain",
              "type": ["int", "null"]
            },
            {
              "name": "buy",
              "type": ["int", "null"]
            },
            {
              "name": "sell",
              "type": ["int", "null"]
            }
          ]
        }
      }
    },
    {
      "name": "Reactions",
      "type": {
        "type": "array",
        "items": {
          "name": "Reactions_record",
          "type": "record",
          "fields": [
            {
              "name": "buy",
              "type": "int"
            },
            {
              "name": "sell",
              "type": "int"
            }
          ]
        }
      }
    }
  ]
}

请注意，操作现在包括所有字段。如果您需要帮助将 Json 转换为 Avro 架构，请使用此 schema generator。

PS：如果您需要更多关于如何控制每次合并的记录数的信息，请click here。