一种方法是执行一个查询，为多值字段orders生成JSON格式的输出。一个使用postgres的示例如下所示：
首先，你会有一个SQL查询文件employees-logstash.sql，内容如下：

WITH json_employee_orders as (
  select
    employee_id,
    JSON_AGG(JSON_BUILD_OBJECT('order_id', order_id, 'product_name', product_name) as orders
  from employee_orders
  group by employee_id
)

SELECT
  e.id, e.last_name, e.first_name, ...,
  jeo.orders::text as orders -- we use ::text because logstash doesn't recognize postgres native JSON type columns
FROM employee e
LEFT JOIN json_employee_orders jeo ON jeo.employee_id = e.id

字符串
然后你的logstash.conf会有这样的内容：

input {
  jdbc {
    jdbc_driver_library => "${PWD}/postgresql-42.6.0.jar"
    jdbc_driver_class => "org.postgresql.Driver"
    jdbc_connection_string => "jdbc:postgresql://${DATABASE_HOST}/${DATABASE_NAME}"
    jdbc_user => "${DATABASE_USER}"
    jdbc_password => "${DATABASE_PASSWORD}"
    statement_filepath => "${PWD}/employees-logstash.sql"
  }
}

filter {
  json {
    source => "orders"
    target => "orders"
  }
}

output {
  opensearch {
    hosts => ["${OPENSEARCH_URL}"]
    user => "${OPENSEARCH_USER}"
    password => "${OPENSEARCH_PASSWORD}"
    index => "employees"
    document_id => "%{[id]}"
  }
}

型