在 Apache NiFi 中使用 AvroSchemaRegistry

Question

我有 5 个不同的 CSVReader 控制器服务。除了schema text（因为不同的header）和1个CSVRecordSetWriter之外，它们的配置是一样的。

我只想留下一个CSVReader 并动态设置schema text。我读到了AvroSchemaRegistry，但我并不清楚如何使用它。

我是否应该创建 5 个不同的 AvroSchemaRegistry 控制器，具有 2 个属性：name 和 value？ F.e.我想输入以下架构：

{
    "type": "record",
    "name": "campaigns",
    "namespace": "common",
    "fields": [
        {"name": "campaign_name", "type": "string"},
        {"name": "campaign_id", "type": "long"},
        {"name": "date", "type" : {"type": "int", "logicalType" : "date"}}
    ]
}

我应该创建AvroSchemaRegistry：

• name = 广告系列；
• value = 上面的完整代码。

对于另一个架构，我应该创建另一个具有另一个属性 name 和 value 的 AvroSchemaResgitry 控制器？

之后，如何配置CSVReader 和CSVRecordSetWriter 以使用这些模式？最后，我应该如何处理流文件？添加其他属性？什么样的？

Answer 1

像这样配置您的流程（根据您的要求进行更改），

1. UpdateAttribute 用于派生/硬编码流文件特定架构的配置-

1. ValidateRecord 配置使用通用 csv 阅读器并动态传递架构 -

1. CSVReader 控制器服务使用动态传递的模式并设置模式访问策略-

如果您希望使用 NiFi 支持的架构注册表，则将所有架构放入注册表并设置 schema.name 和 access strategy 属性以供记录读取器/写入器从注册表访问架构，但首先您需要添加/配置架构控制器服务中的注册表提供程序。

更新：

HortonworksSchemaRegistry 的示例 SchemaRegistry 控制器服务配置（ConfluentSchemaRegistry 和 AvroSchemaRegistry 具有或多或少相同的属性）。重要的是执行查找操作时注册表 API 期望的参数，因此我们只需要从 Reader/Writer 控制器服务Schema Name 属性传递相同的值，请refer to this 了解一下。

架构注册表网址：http://example.com:7788/api/v1