flink sql展开列并添加索引

Question

使用flink sql有没有办法将列表解析为多行并在列表中添加索引？

以下是输入示例:

{
    "uid":"123",
    "imp_content":[
        {"key":"#a#"},
        {"key":"#b#","search_flag":"5"},
        {"key":"#c#","adid":150690},
        {"key":"#d#","search_flag":"0"},
        {"key":"#e#","search_flag":"0"},
        {"key":"#f#","search_flag":"0"},
        {"key":"#g#","search_flag":"5"},
        {"key":"#x#","search_flag":"5"},
        {"key":"#i#","search_flag":"3"}
    ]
}

以下是预期结果：

谢啦

Answer 1

下面是一个执行类似操作的示例。给定此表：

CREATE TABLE `flink_commits` (
  `author` VARCHAR(2147483647),
  `authorDate` TIMESTAMP(3),
  `authorEmail` VARCHAR(2147483647),
  `commitDate` TIMESTAMP(3),
  `committer` VARCHAR(2147483647),
  `committerEmail` VARCHAR(2147483647),
  `filesChanged` ARRAY<ROW<`filename` VARCHAR(2147483647), `linesAdded` INT, `linesChanged` INT, `linesRemoved` INT>>,
  `sha1` VARCHAR(2147483647),
  `shortInfo` VARCHAR(2147483647),
  WATERMARK FOR `commitDate` AS `commitDate` - INTERVAL '1' DAY
)
COMMENT 'Commits on the master branch of github.com/apache/flink'
WITH (
  'connector' = 'kafka',
  ...
);

您可以这样做:

SELECT
  sha1, filename, added, changed, removed
FROM
  flink_commits
CROSS JOIN UNNEST(filesChanged) AS files (filename, added, changed, removed);