使用Logstash从PostgreSQL同步为Elasticsearch文档添加数组字段

现有配置核心问题 #

• aggregate过滤器仅初始化了persons空数组，未将查询到的人员信息追加到数组中
• 未清理聚合后事件的冗余字段，可能导致无关字段被写入ES覆盖原有内容
• 未配置聚合插件所需的单线程运行环境，多线程下聚合逻辑会失效
• 部分配置语法不规范，容易引发解析错误

修正后的完整配置 #

input {
    jdbc {
        jdbc_connection_string => "jdbc:postgresql://localhost:5432/app"
        jdbc_user => "postgres"
        jdbc_password => "admin"
        jdbc_driver_class => "org.postgresql.Driver"
        jdbc_driver_library => "postgresql-42.2.21.jar"
        # 给字段加别名，避免后续字段名混乱
        statement => "select id::text, appointment_id::text,trim(firstname) as firstname,trim(lastname) as lastname from appointments_persons order by created_at"
        # 增量同步可选配置，按需开启
        # tracking_column => "created_at"
        # use_column_value => true
    }
}
filter {
    aggregate {
        task_id => "%{appointment_id}"
        code => "
            map['persons'] ||= []
            # 将当前行的人员信息追加到聚合数组中
            map['persons'] << {
                'firstname' => event.get('firstname'),
                'lastname' => event.get('lastname')
            }
            # 标记单条行事件已处理，不需要单独输出
            event.cancel()
        "
        push_map_as_event_on_timeout => true
        timeout_task_id_field => "appointment_id"
        timeout => 10
    }
    # 清理冗余字段，只保留需要更新的persons字段和用于关联文档的appointment_id
    mutate {
        remove_field => ["@timestamp", "@version", "id", "firstname", "lastname"]
    }
}
output {
    #stdout { codec => json_lines }
    elasticsearch {
        hosts => "127.0.0.1"
        index => "appointments"
        # 明确指定更新操作，字符串加引号避免语法错误
        action => "update"
        # 仅更新传入的字段，不会覆盖文档原有其他内容
        doc_as_upsert => true
        document_id => "%{appointment_id}"
    }
}

额外注意事项 #

• aggregate插件要求Logstash管道必须为单线程运行，否则同一个appointment_id的记录会被分发到不同线程导致聚合失败，启动时需要加参数logstash -w 1 -f 你的配置文件路径，或者在logstash.yml中设置pipeline.workers: 1。

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