Logstash聚合报错：TypeError: 无法将String隐式转为Integer

错误描述 #

使用Logstash的aggregate过滤器时遇到类型转换错误：

TypeError: no implicit conversion of String into Integer

过滤器接收来自SQL的多类型输入字段（字符串、整数、小数），配置代码如下：

filter {
    aggregate {
        task_id => "%{providerid}"
       code => "
            map['providerid'] ||= event.get('providerid')
            map['pricingbytrade'] ||= []
            kpishistorical = {}
            kpiscurrent = {}
            hash = event.to_hash
            hash.each { |key,value|
                if key != 'providerid' && key != 'cbsaid' && key != 'primarytrade' && key != 'cbsapricingscore' && key != 'cbsaname' && key != 'scoreyear' && key != 'type'  && !key.include?('@') && key != 'historicalcbsapricingscore' && !key.start_with?('current')
                    if value.is_a? Numeric
                        value = value.to_f
                    end
                    kpishistorical.store(key, value)
                end
                if key != 'providerid' && key != 'cbsaid' && key != 'primarytrade' && key != 'cbsapricingscore' && key != 'cbsaname' && key != 'scoreyear' && key != 'type'  && !key.include?('@') && !key.start_with?('historical')
                    if value.is_a? Numeric
                        value = value.to_f
                    end
                    kpiscurrent.store(key, value)
                end
            }

           map['pricingbytrade'].find {|m| m['name'] == event.get('primarytrade')}['pricingbycbsa'] <<
                    {
                        'cbsaid' => event.get('cbsaid'),
                        'cbsaname' => event.get('cbsaname'),
                        'pricingscore' => event.get('cbsapricingscore'),
                        'kpis' => kpiscurrent,
                        'historical' => [
                            {
                                'scoremonth' => event.get('scoremonth'),
                                'scoreyear' => event.get('scoreyear'),
                                'pricingscore' => event.get('historicalcbsapricingscore'),
                                'kpis' => kpishistorical
                            }
                        ]
                    }
     event.cancel()
    "
  }
}

输入字段说明：cbsapricingscore和historicalcbsapricingscore为小数类型，cbsaid为整数类型。

错误原因 #

报错的核心原因是：当处理某个primarytrade的第一条数据时，map['pricingbytrade']中还没有对应名称的条目，find方法返回nil，此时尝试对nil调用['pricingbycbsa']，Ruby会将字符串索引尝试转换为整数（数组索引要求），从而触发类型转换错误。

另外，SQL输入的数值字段可能被Logstash解析为字符串类型，若后续未显式转换，可能引发潜在的类型不一致问题。

修复方案 #

1. 先检查并初始化缺失的trade条目：在访问pricingbycbsa之前，判断是否找到对应trade，若未找到则创建新条目并加入数组。
2. 显式转换字段类型：对SQL输入的数值字段进行类型转换，确保后续处理类型一致。

修复后的配置代码：

filter {
    aggregate {
        task_id => "%{providerid}"
        code => "
            map['providerid'] ||= event.get('providerid')
            map['pricingbytrade'] ||= []
            kpishistorical = {}
            kpiscurrent = {}
            hash = event.to_hash
            
            hash.each { |key,value|
                if key != 'providerid' && key != 'cbsaid' && key != 'primarytrade' && key != 'cbsapricingscore' && key != 'cbsaname' && key != 'scoreyear' && key != 'type'  && !key.include?('@') && key != 'historicalcbsapricingscore' && !key.start_with?('current')
                    # 统一处理数值类型，字符串转浮点
                    value = value.is_a?(Numeric) ? value.to_f : (value.to_f rescue value)
                    kpishistorical.store(key, value)
                end
                if key != 'providerid' && key != 'cbsaid' && key != 'primarytrade' && key != 'cbsapricingscore' && key != 'cbsaname' && key != 'scoreyear' && key != 'type'  && !key.include?('@') && !key.start_with?('historical')
                    # 统一处理数值类型，字符串转浮点
                    value = value.is_a?(Numeric) ? value.to_f : (value.to_f rescue value)
                    kpiscurrent.store(key, value)
                end
            }

            # 查找或创建对应的trade条目
            primary_trade = event.get('primarytrade')
            trade_entry = map['pricingbytrade'].find {|m| m['name'] == primary_trade}
            
            unless trade_entry
                # 初始化新的trade条目
                trade_entry = {
                    'name' => primary_trade,
                    'pricingbycbsa' => []
                }
                map['pricingbytrade'] << trade_entry
            end

            # 向pricingbycbsa数组添加数据
            trade_entry['pricingbycbsa'] << {
                'cbsaid' => event.get('cbsaid').to_i, # 显式转整数
                'cbsaname' => event.get('cbsaname'),
                'pricingscore' => event.get('cbsapricingscore').to_f, # 显式转浮点
                'kpis' => kpiscurrent,
                'historical' => [
                    {
                        'scoremonth' => event.get('scoremonth'),
                        'scoreyear' => event.get('scoreyear').to_i, # 显式转整数
                        'pricingscore' => event.get('historicalcbsapricingscore').to_f, # 显式转浮点
                        'kpis' => kpishistorical
                    }
                ]
            }
            
            event.cancel()
        "
    }
}

关键修改点 #

• 增加了trade条目的检查与初始化逻辑，避免nil引用导致的类型转换错误。
• 对数值型字段（cbsaid、cbsapricingscore等）进行显式类型转换，确保类型一致性。
• 优化了数值类型处理逻辑，使用rescue避免非数值字符串转浮点时报错。

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