apache-kafka - 如何使用 Kafka Connect AVRO 的 Schema 注册表

Question

我最近开始研究 Kafka 和 Kafka connect 并进行了一些初始设置。但想在模式注册表部分探索更多。

我的模式注册现在开始了，我应该做什么。

我有一个 AVRO 模式存储在 avro_schema.avsc 中。

这是架构

{
  "name": "FSP-AUDIT-EVENT",
  "type": "record",
  "namespace": "com.acme.avro",
  "fields": [
    {
      "name": "ID",
      "type": "string"
    },
    {
      "name": "VERSION",
      "type": "int"
    },
    {
      "name": "ACTION_TYPE",
      "type": "string"
    },
    {
      "name": "EVENT_TYPE",
      "type": "string"
    },
    {
      "name": "CLIENT_ID",
      "type": "string"
    },
    {
      "name": "DETAILS",
      "type": "string"
    },
    {
      "name": "OBJECT_TYPE",
      "type": "string"
    },
    {
      "name": "UTC_DATE_TIME",
      "type": "long"
    },
    {
      "name": "POINT_IN_TIME_PRECISION",
      "type": "string"
    },
    {
      "name": "TIME_ZONE",
      "type": "string"
    },
    {
      "name": "TIMELINE_PRECISION",
      "type": "string"
    },
    {
      "name": "AUDIT_EVENT_TO_UTC_DT",
      "type": [
        "string",
        "null"
      ]
    },
    {
      "name": "AUDIT_EVENT_TO_DATE_PITP",
      "type": "string"
    },
    {
      "name": "AUDIT_EVENT_TO_DATE_TZ",
      "type": "string"
    },
    {
      "name": "AUDIT_EVENT_TO_DATE_TP",
      "type": "string"
    },
    {
      "name": "GROUP_ID",
      "type": "string"
    },
    {
      "name": "OBJECT_DISPLAY_NAME",
      "type": "string"
    },
    {
      "name": "OBJECT_ID",
      "type": [
        "string",
        "null"
      ]
    },
    {
      "name": "USER_DISPLAY_NAME",
      "type": [
        "string",
        "null"
      ]
    },
    {
      "name": "USER_ID",
      "type": "string"
    },
    {
      "name": "PARENT_EVENT_ID",
      "type": [
        "string",
        "null"
      ]
    },
    {
      "name": "NOTES",
      "type": [
        "string",
        "null"
      ]
    },
    {
      "name": "SUMMARY",
      "type": [
        "string",
        "null"
      ]
    }
  ]
}

我的模式是否有效。我从 JSON 在线转换了吗？我应该在哪里保存这个模式文件的位置我不确定。请指导我执行以下步骤.我从 Lambda 函数和 JDBC 源发送记录。

那么基本上我如何执行 AVRO 架构和测试？我是否必须更改 avro-consumer 属性文件中的任何内容？

或者这是注册模式的正确方法

   ./bin/kafka-avro-console-producer \
                 --broker-list b-3.**:9092,b-**:9092,b-**:9092 --topic AVRO-AUDIT_EVENT \
                 --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string"}]}'





curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json"     --data '{"schema" : "{\"type\":\"struct\",\"fields\":[{\"type\":\"string\",\"optional\":false,\"field\":\"ID\"},{\"type\":\"string\",\"optional\":true,\"field\":\"VERSION\"},{\"type\":\"string\",\"optional\":true,\"field\":\"ACTION_TYPE\"},{\"type\":\"string\",\"optional\":true,\"field\":\"EVENT_TYPE\"},{\"type\":\"string\",\"optional\":true,\"field\":\"CLIENT_ID\"},{\"type\":\"string\",\"optional\":true,\"field\":\"DETAILS\"},{\"type\":\"string\",\"optional\":true,\"field\":\"OBJECT_TYPE\"},{\"type\":\"string\",\"optional\":true,\"field\":\"UTC_DATE_TIME\"},{\"type\":\"string\",\"optional\":true,\"field\":\"POINT_IN_TIME_PRECISION\"},{\"type\":\"string\",\"optional\":true,\"field\":\"TIME_ZONE\"},{\"type\":\"string\",\"optional\":true,\"field\":\"TIMELINE_PRECISION\"},{\"type\":\"string\",\"optional\":true,\"field\":\"GROUP_ID\"},{\"type\":\"string\",\"optional\":true,\"field\":\"OBJECT_DISPLAY_NAME\"},{\"type\":\"string\",\"optional\":true,\"field\":\"OBJECT_ID\"},{\"type\":\"string\",\"optional\":true,\"field\":\"USER_DISPLAY_NAME\"},{\"type\":\"string\",\"optional\":true,\"field\":\"USER_ID\"},{\"type\":\"string\",\"optional\":true,\"field\":\"PARENT_EVENT_ID\"},{\"type\":\"string\",\"optional\":true,\"field\":\"NOTES\"},{\"type\":\"string\",\"optional\":true,\"field\":\"SUMMARY\"},{\"type\":\"string\",\"optional\":true,\"field\":\"AUDIT_EVENT_TO_UTC_DT\"},{\"type\":\"string\",\"optional\":true,\"field\":\"AUDIT_EVENT_TO_DATE_PITP\"},{\"type\":\"string\",\"optional\":true,\"field\":\"AUDIT_EVENT_TO_DATE_TZ\"},{\"type\":\"string\",\"optional\":true,\"field\":\"AUDIT_EVENT_TO_DATE_TP\"}],\"optional\":false,\"name\":\"test\"}"}' http://localhost:8081/subjects/view/versions

接下来我要做什么

但是当我尝试查看我的模式时，我只得到了下面的内容

curl --silent -X GET http://localhost:8081/subjects/AVRO-AUDIT-EVENT/versions/latest

这是结果

{"subject":"AVRO-AUDIT-EVENT","version":1,"id":161,"schema":"{\"type\":\"string\",\"optional\":false}"}

为什么我看不到我的完整注册模式

还有当我尝试删除模式时

我得到以下错误

{"error_code":405,"message":"HTTP 405 Method Not Allowed"

我不确定我的架构是否已正确注册。

请帮帮我。提前致谢

Answer 1

is my schema valid

您可以使用 Registry 的 REST API 尝试提交并查看...

where should i keep this schema file location i am not sure about

目前还不清楚您是如何发送消息的...

如果您实际编写了 Kafka 生产者代码，则将其存储在您的代码中(作为字符串)或作为资源文件。如果使用 Java，您可以改用 SchemaBuilder 类来创建 Schema 对象

如果您还没有使用 Avro Schema 和序列化器，您需要重写您的生产者

If we create AVRO schema will it work for Json as well .

Avro 是一种二进制格式，但它有一个 JSONDecoder。

what should be the URL for our AVRO schema properties file ?

一旦您弄清楚如何启动它，它就需要是您的架构注册表的 IP。 (使用 schema-registry-start)

Do i have to change anything in avro-consumer properties file ?

您需要使用 Avro 解串器

is this correct way to register schema

.> /bin/kafka-avro-console-producer\

不完全是。这就是您使用模式生成消息的方式(并且您需要使用正确的模式)。您还必须提供 --property schema.registry.url

您使用注册表的 REST API 来注册和验证模式