Logstash Avro插件传递long类型值异常的解决方法咨询

这个问题我之前排查过类似的，核心就是你推测的Logstash Avro插件的64位整数溢出问题——尤其是你用的6.1.2这种比较老的版本，它的Avro codec底层依赖的库对long类型的处理有bug，导致解析出来的值异常。下面给你几个不用把long转成字符串的可行解决思路：

• 升级Logstash及Avro插件版本
  老版本的Logstash（比如6.x系列）的Avro codec存在已知的64位整数解析bug，升级到7.x及以上版本的Logstash，同时更新Avro插件是最彻底的解决办法。你可以通过以下命令更新插件：

  bin/logstash-plugin update logstash-codec-avro
  

  新版本的插件修复了long类型的溢出问题，能正确解析Java端发送的64位整数。

• 用Ruby Filter手动修复溢出值（临时 workaround）
  如果暂时无法升级版本，可以在Logstash的Filter阶段通过Ruby代码还原溢出的long值。原理是把被错误当成32位int解析的值，重新计算还原为原本的64位long：

  filter {
    ruby {
      code => "
        # 修复scoringId字段的溢出问题
        if event.get('scoringId')
          raw_value = event.get('scoringId').to_i
          # 处理32位int溢出逻辑，还原64位long
          fixed_value = raw_value >= 0 ? raw_value : raw_value + 2**32
          event.set('scoringId', fixed_value)
        end
        # 同理处理时间戳字段，替换成你的时间戳字段名
        if event.get('your_timestamp_field')
          raw_ts = event.get('your_timestamp_field').to_i
          fixed_ts = raw_ts >= 0 ? raw_ts : raw_ts + 2**32
          event.set('your_timestamp_field', fixed_ts)
        end
      "
    }
  }
  

  这个方法适用于大部分场景，只要你的long值没有超出64位整数范围，都能正确还原。

• 替换官方Avro Codec，用原生Avro库解析
  如果上述方法都无效，可以绕开官方的Avro codec，直接在Filter阶段用Avro原生库解析原始字节：

  1. 先修改Kafka Input的Codec为plain，拿到原始二进制数据：

     input {
       kafka {
         bootstrap_servers => 'kafkaserver:9092'
         topics => ["bart.vector"]
         codec => plain
         client_id => "logstash-vector-tracking"
       }
     }
     

  2. 安装Ruby的Avro库：gem install avro
  3. 在Filter阶段用Ruby代码解析Avro数据：

     filter {
       ruby {
         code => "
           require 'avro'
           # 加载Avro Schema
           schema_file = File.open('C:\logstash-6.1.2\vectorInfoDWH.avsc', 'r')
           schema = Avro::Schema.parse(schema_file.read)
           # 解析原始二进制数据
           decoder = Avro::IO::BinaryDecoder.new(StringIO.new(event.get('message')))
           reader = Avro::IO::DatumReader.new(schema)
           parsed_data = reader.read(decoder)
           # 将解析后的字段写入事件
           parsed_data.each do |key, value|
             event.set(key, value)
           end
           # 移除原始message字段
           event.remove('message')
         "
       }
     }
     

  这种方法直接使用Avro官方的Ruby库解析，完全避免了Logstash插件的bug，兼容性最好，但需要额外安装依赖。

• 最后再确认Schema和Java端的序列化
  虽然你说Java端的序列化是正确的，但可以再快速核对下：

  • 确认Avro Schema里的字段类型确实是"type": "long"，没有误写成int
  • 确认Java生成的DTO中对应字段是long类型（而非int），序列化时没有做不必要的类型转换

内容的提问来源于stack exchange，提问作者the_kaba