发布数据损坏时跳过 kafka 中的 sink 步骤答案

【问题标题】：skip the sink step in kafka when posted data corrupted发布数据损坏时跳过 kafka 中的 sink 步骤
【发布时间】：2016-09-02 07:56:15
【问题描述】：

在java服务器端经过一些处理后，我通过restful webservice将日志数据（json格式）从服务器发布到kafka。

在 hdfs 方面我的接收器类型是 avro。因此，为了将 json（源）解析为 avro（目标），我使用的是 morphline 和 avro 模式。

如果发布的数据不适合 morphline 或 avro 模式，通常我会收到以下错误，

原因：com.fasterxml.jackson.core.JsonParseException：非法不带引号的字符（（CTRL-CHAR，代码 10））：必须使用转义要包含在字符串值中的反斜杠

如果我得到这个一次，偏移量就不再移动了。简而言之，如果 kafka 只收到一次此错误，它就不能再接收已发布的数据了。

为了避免这个错误，我想有两种解决方案。第一个是在服务器端为大数据端使用的 avro 模式编写 json 验证器。我更喜欢的第二种方法是跳过并且不接收未格式化为请求的 avro 模式的日志数据。但是在跳过损坏的数据后，如果 kafka 获得合适的数据，它应该下沉它。

我认为如果我在flume或kafka配置文件中添加一些参数是可能的。那么，当发布的数据不适合请求的模式或请求的 morphline 时，如何跳过接收器步骤？

【问题讨论】：

标签： hadoop hdfs apache-kafka avro flume

【解决方案1】：

我解决了吗啉方面的问题，

像这样在 morphline 中添加了 try-catch 代码块

morphlines: [
  {
    id: convertJsonToAvro
    importCommands: [ "org.kitesdk.**" ]
    commands: [
       {
         tryRules {
              catchExceptions : true
           rules : [
             {
               commands : [
                 # save initial state
                 { readJson {} }
                # extract JSON objects into fields
              { extractJsonPaths {
                flatten: true
                paths: {
            PROJECT_NAME: /PROJECT_NAME
            WSDL_NAME: /WSDL_NAME
            ....
            ....
            ....
            MESSAGE_OUT: /MESSAGE_OUT
        }
      } }
      # convert the extracted fields to an avro object
      # described by the schema in this field
      { toAvro {
        schemaFile:/u0x/myPath/myAvroSchema.avsc
      } }
      # serialize the object as avro
      { writeAvroToByteArray: {
        format: containerlessBinary
              } }
           ]
         }
         {
          commands : [
            { logWarn { format : "Ignoring record with unsupported input format in myLogService: {}", args : ["@{}"] } }
            { dropRecord {} }    
            ]
         }
       ]
     }   
    }    
   ]
  }
]

在tryRules 中，我强制代码捕获所有异常。

在rules:你可以写"command:"块任何你想要的，如果其中一个抛出除了最后一个命令块之外的异常，最后一个命令将运行。请记住，最后一个是“catch”。我的意思是，如果第一个命令块失败，最后一个（第二个）命令将运行。如果第一个命令完美运行，最后一个命令将不起作用，因为最后一个命令块像一个 catch 块一样工作。

因此，当代码 readJson {} 在第一个命令块中失败时，它会抛出异常，最后一个命令（catch 块）会处理它，因此它不会尝试在 kafka 主题中接收当前数据，因为它将运行 dropRecord {}。

详细文档可以访问kitesdk。

【讨论】：