mysql 中 一对多的数据源如何 sink到 ES

[promise2mm]

• 场景：商品 + sku （一对多），一个商品可以关联多个sku
• mysql表
  • goods(id, name)
  • sku(sku_id, goods_id, price)
• ES sink 索引结构如下

  {
         "id" : 1,
         "name" : "测试goods name",
         "skus" : [
           {
             "sku_id" : 1001,
             "goods_id" : 1,
             "price" : 100
           },
           {
             "sku_id" : 1002,
             "goods_id" : 1,
             "price" : 200
           }
         ]
       }
  

• 问题：如何实现这个场景诉求？针对于skus这个es字段，flink sql 中支持这样的对象数组吗？

[promise2mm]

直接用Flink-SQL能否实现？尝试过定义一个这样的表结构

CREATE TABLE goods_detail_test (
 		goods_id INT,
 		skus ARRAY<ROW(sku_id INT, goods_id INT, price DECIMAL(10,5))>,
 		name STRING,
 		PRIMARY KEY(goods_id) NOT ENFORCED
 	) WITH (
 		'connector' = 'elasticsearch-7', 'hosts' = 'http://localhost:9200',
 		'index' = 'goods_detail_test' 
);

但是insert语句不知道如何处理skus这个array对象了。

[ashulin]

1. 可以通过ES的父子文档，Flink写对应索引，不定义复杂结构；ES在这个场景下性能不是很好；
2. 通过Flink SQL JOIN后多列转行，即你说的方案，这个Flink是可行的。可能需要自己实现一个UDF方法；

[promise2mm]

嗯，看起来得考虑自己实现了
第一种方案使用父子结构，性能太差了
thx!