我现在使用logstash创建一个机制，将postgres的数据发送到elasticsearch。
我有一个postgres的父子表。
我现在想把它放在一个elasticsearch文档中。
数据库定义如下。

CREATE TABLE tenant(
  tenant_id        SERIAL PRIMARY KEY,
  name             VARCHAR(100)
);
INSERT INTO tenant(name) VALUES('gakus company');
INSERT INTO tenant(name) VALUES('yamada company');
INSERT INTO tenant(name) VALUES('hanako company');
CREATE TABLE card(
  card_id        SERIAL PRIMARY KEY,
  tenant_id         INTEGER,
  title             VARCHAR(20)
);
INSERT INTO card(tenant_id, title) VALUES(1,'card1');
INSERT INTO card(tenant_id, title) VALUES(1,'card2');
INSERT INTO card(tenant_id, title) VALUES(2,'card1');
INSERT INTO card(tenant_id, title) VALUES(2,'card2');
INSERT INTO card(tenant_id, title) VALUES(3,'card1');
INSERT INTO card(tenant_id, title) VALUES(3,'card2');
CREATE TABLE profile(
  profile_id        SERIAL PRIMARY KEY,
  card_id           INTEGER,
  name              VARCHAR(100),
  age               INTEGER
);
INSERT INTO profile(card_id, name, age) VALUES(1,'gaku', 30);
INSERT INTO profile(card_id, name, age) VALUES(3,'yamada', 45);
INSERT INTO profile(card_id, name, age) VALUES(5,'hanako', 70);

我创建了下面的logstash管道。

input {
  jdbc {
    jdbc_driver_library => "/usr/share/java/postgresql-jdbc.jar"
    jdbc_driver_class => "org.postgresql.Driver"
    jdbc_connection_string => "jdbc:postgresql://db:5432/database"
    jdbc_user => "user"
    jdbc_password => "password"
    jdbc_default_timezone => "Asia/Tokyo"
    schedule => "*/5 * * * * *" # 同期タイミングを指定
    statement => "SELECT * FROM tenant"
  }
}

filter {
  mutate {
    copy => { "tenant_id" => "[@metadata][_id]"} # ここでidをcopy
    remove_field => ["id", "@version", "unix_ts_in_secs"] # elasticsearch上の不要情報の削除
  }

  jdbc_streaming {
    jdbc_driver_library => "/usr/share/java/postgresql-jdbc.jar"
    jdbc_driver_class => "org.postgresql.Driver"
    jdbc_connection_string => "jdbc:postgresql://db:5432/database"
    jdbc_user => "user"
    jdbc_password => "password"
    statement => "select * from card WHERE tenant_id = :tenant_id"
    parameters => {"tenant_id" => "tenant_id"}
    target => "cards"
  }
}

output {
  elasticsearch {
    hosts => ["elasticsearch"]
    index => "searchlist"
    document_id => "searchlist-%{[@metadata][_id]}" # document_idをテーブルのidと一致させる
  }
}

租客→ 卡已成功嵌套并存储在elasticsearch中。
但是，我不知道如何在ElastCasearch中使用租户的三阶段结构保存它→ 卡片→ 轮廓。
您知道如何在多层配置中保持嵌套并将其保存在elasticsearch文档中吗？
如果你有什么好主意，请告诉我。
谢谢你！