-- 假设我们有一个MySQL表，包含一个JSON类型的字段用于存储数组数据
CREATE TABLE `my_table` (
`id` INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
`data` JSON
);
-- 假设我们要同步的JSON数据中包含一个数组字段 `my_array`
-- 我们需要将这个数组字段展开为多行格式，以便每个数组元素对应一行
-- 使用JSON_EXTRACT函数和RECURSIVE CTE进行展开
-- 创建一个临时的表，用于存储同步过程中的元信息
CREATE TEMPORARY TABLE `es_metadata` (
`id` INT PRIMARY KEY,
`version` VARCHAR(10),
`data` JSON
);
-- 插入一条示例数据，其中`my_array`包含两个元素
INSERT INTO `my_table` (`data`) VALUES ('{"my_array": ["elem1", "elem2"]}');
-- 使用RECURSIVE CTE来展开JSON数组
WITH RECURSIVE cte (id, version, data, path, value) AS (
SELECT
t.id,
'v1',
t.data,
CAST('$' AS JSON),
JSON_EXTRACT(t.data, '$')
FROM
my_table t
UNION ALL
SELECT
cte.id,
cte.version,
cte.data,
JSON_UNQUOTE(JSON_EXTRACT(cte.path, '$[0]')) AS path,
JSON_EXTRACT(cte.value, JSON_UNQUOTE(JSON_EXTRACT(cte.path, '$[0]'))) AS value
FROM
cte
WHERE
JSON_TYPE(JSON_EXTRACT(cte.value, JSON_UNQUOTE(JSON_EXTRACT(cte.path, '$[0]')))) = 'ARRAY'
UNION ALL
SELECT
cte.id,
cte.version,
cte.data,
CONCAT(cte.path, '[', JSON_UNQUOTE(JSON_EXTRACT(cte.path, '$[0]')), '].[', cte.idx, ']') AS path,
JSON_EXTRACT(JSON_EXTRACT(cte.value, JSON_UNQUOTE(JSON_EXTRACT(cte.path, '$[0]'))), cte.idx) AS value
FROM
cte
JOIN (
SELECT
0 AS idx
UNION ALL
SELECT
idx + 1 AS idx
FROM
cte
WHERE
idx < JSON_LENGTH(JSON_EXTRACT(cte.value, JSON_UNQUOTE(JSON_EXTRACT(cte.path, '$[0]'))))
) AS indexes ON 1
)
-- 将展开的数组数据插入到Elasticsearch
INSERT INTO `es_metadata` (id, version, data)
SELECT
t.id,
t.version,
JSON_OBJECT(
'my_array',
JSON_ARRAYAGG(IF(JSON_TYPE(cte.value) = 'ARRAY', cte.value, cte.value))
)
FROM
cte
RIGHT JOIN
my_table t ON cte.id = t.id
GROUP BY
t.id;
-- 注意：这里的INSERT INTO `es_metadata` 语句是假设的，因为实际的Elasticsearch同步逻辑会根据版本和数据类型进行处理。
-- 这个例子展示了如何将MySQL中的JSON数组数据转换为多行格式，以便于同步到Elasticsearch。这段代码展示了如何将一个JSON数组字段展开为多行，并且如何使用RECURSIVE CTE来处理嵌套的JSON数组。这