Flink sql 如何行转列?

2021-03-26 19:08发布

1条回答
studentaaa
2楼 · 2021-04-18 10:38

在 SQL 任务里面经常会遇到一列转多行的需求,今天就来总结一下在 Flink SQL 里面如何实现列转行的,先来看下面的一个具体案例.


需求

原始数据格式如下:


name data

JasonLee [{"content_type":"flink","url":"111"},{"content_type":"spark","url":"222"},{"content_type":"hadoop","url":"333"}]

data 格式化

{

 "name": "JasonLee",

 "data": [{

   "content_type": "flink",

   "url": "111"

  }, {

   "content_type": "spark",

   "url": "222"

  },

  {

   "content_type": "hadoop",

   "url": "333"

  }

 ]

}

现在希望得到的数据格式是这样的:


name content_type url

JasonLee flink 111

JasonLee spark 222

JasonLee hadoop 333

这是一个典型的列转行或者一行转多行的场景,需要将 data 列进行拆分成为多行多列,下面介绍两种实现方式.


使用 Flink 自带的 unnest 函数解析

使用自定义 UDTF 函数解析

建表 DDL

CREATE TABLE kafka_table (

name string,

`data` ARRAY>

)

WITH (

    'connector' = 'kafka', -- 使用 kafka connector

    'topic' = 'test',

    'properties.bootstrap.servers' = 'master:9092,storm1:9092,storm2:9092',  -- broker连接信息

    'properties.group.id' = 'jason_flink_test', -- 消费kafka的group_id

    'scan.startup.mode' = 'latest-offset',  -- 读取数据的位置

    'format' = 'json',  -- 数据源格式为 json

    'json.fail-on-missing-field' = 'false', -- 字段丢失任务不失败

    'json.ignore-parse-errors' = 'true'  -- 解析失败跳过

)

这里在定义 data 字段类型的时候需要定义为 ARRAY 类型,因为 unnest 函数需要一个数组类型的参数.


unnest 解析

select name,content_type,url

from kafka_table CROSS JOIN UNNEST(`data`) AS t (content_type,url)

select name,content_type,url

from kafka_table, UNNEST(`data`) AS t (content_type,url)

select name,content_type,url

from kafka_table left join UNNEST(`data`) AS t (content_type,url) on true

自定义 UDTF 解析

自定义表值函数(UDTF),自定义表值函数,将 0 个、1 个或多个标量值作为输入参数(可以是变长参数)。与自定义的标量函数类似,但与标量函数不同。表值函数可以返回任意数量的行作为输出,而不仅是 1 个值。返回的行可以由 1 个或多个列组成。调用一次函数输出多行或多列数据。必须继承 TableFunction 基类,并实现一个或者多个名为 eval 的方法, 在使用 UDTF 时,需要带上 LATERAL TABLE两个关键字.


@FunctionHint(output = @DataTypeHint("ROW"))

public class ParserJsonArrayTest extends TableFunction {

 

    private static final Logger log = Logger.getLogger(ParserJsonArrayTest.class);

 

    public void eval(String value) {

        try {

            JSONArray snapshots = JSONArray.parseArray(value);

            Iterator iterator = snapshots.iterator();

            while (iterator.hasNext()) {

                JSONObject jsonObject = (JSONObject) iterator.next();

                String content_type = jsonObject.getString("content_type");

                String url = jsonObject.getString("url");

                collect(Row.of(content_type,url));

            }

        } catch (Exception e) {

            log.error("parser json failed :" + e.getMessage());

        }

    }

}

自定义 UDTF 解析的时候,就不需要把 data 字段定义成 ARRAY 类型了,直接定义成 STRING 类型就可以了,并且这种方式会更加的灵活,比如还需要过滤数据或者更复杂的一些操作时都可以在 UDTF 里面完成.


Flink SQL 使用 UDTF

select name,content_type,url

from kafka_table CROSS JOIN lateral TABLE (ParserJsonArrayTest(`data`)) AS t (content_type,url)

select name,content_type,url

from kafka_table, lateral TABLE (ParserJsonArrayTest(`data`)) AS t (content_type,url)

select name,content_type,url

from kafka_table left join lateral TABLE (ParserJsonArrayTest(`data`)) AS t (content_type,url) on true

注意:

unnest 和 自定义 UDTF 函数在使用的时候都有 3 种写法,前面两种写法的效果其实是一样的,第三种写法相当于 left join 的用法.区别在于 CROSS JOIN/INNER JOIN: 对于左侧表的每一行,右侧 UDTF 不输出,则这一行不输出.LEFT JOIN: 对于左侧表的每一行,右侧 UDTF 不输出,则这一行会输出,右侧 UDTF 字段为 null

相关问题推荐

  • 回答 10

    创建test文件夹hadoop fs -mkdir /test

  • 回答 7

    Hadoop的三大核心组件分别是:1、HDFS(Hadoop Distribute File System):hadoop的数据存储工具。2、YARN(Yet Another Resource Negotiator,另一种资源协调者):Hadoop 的资源管理器。3、Hadoop MapReduce:分布式计算框架。HDFS是一个高度容错性的系统,适合部...

  • 回答 18

    hbase依靠HDFS来存储底层数据。Hadoop分布式文件系统(HDFS)为HBase提供了高可靠性的底层存储支持,HBase中的所有数据文件都存储在Hadoop HDFS文件系统上。

  • 回答 24

    HBase分布式数据库具有如下的显著特点:容量大:HBase分布式数据库中的表可以存储成千上万的行和列组成的数据。面向列:HBase是面向列的存储和权限控制,并支持独立检索。列存储,其数据在表中是按照某列存储的,根据数据动态的增加列,并且可以单独对列进行...

  • 回答 19

    解决问题的层面不一样首先,Hadoop和Apache Spark两者都是大数据框架,但是各自存在的目的不尽相同。Hadoop实质上更多是一个分布式数据基础设施: 它将巨大的数据集分派到一个由普通计算机组成的集群中的多个节点进行存储,意味着您不需要购买和维护昂贵的服务...

  • 回答 14

    1、HBase写快读慢,HBase的读取时长通常是几毫秒,而Redis的读取时长通常是几十微秒。性能相差非常大。2、HBase和Redis都支持KV类型。但是Redis支持List、Set等更丰富的类型。3、Redis支持的数据量通常受内存限制,而HBase没有这个限制,可以存储远超内存大小...

  • 回答 15

    列式存储格式是指以列为单位存储数据的数据存储格式,相比于传统的行式存储格式,它具有压缩比高、读I/O少(此处指可避免无意义的读I/O)等优点,目前被广泛应用于各种存储引擎中。对于HBase而言,它并不是一个列式存储引擎,而是列簇式存储引擎,即同一列簇中...

  • 回答 14

    一、简单理解Hadoop是一个大象:一个hadoop集群主要包含三个主要的模块:Mapreduce,hdfs,yarn。mapreduce是一个分离在合并的计算框架,注意他不是一个集群,而是一个编程框架。hdfs是一个分布式文件系统,是一个分布式集群,用于存放数据。yarn集群是负责集群...

  • 回答 12

    01 网络公开数据集02 数据报采集03 网络爬虫04 日志收集05 社会调查06 业务数据集07 埋点采集08 传感器采集09 数据交易平台10 个人数据收集

  • 回答 9

    1 Hadoop 各个目录的解释bin:Hadoop管理脚本和使用脚本所在目录, sbin目录下的脚本都是使用此目录下的脚本实现的。etc:Hadoop的所有配置文件所在的目录,所有hadoop的配置在etc/hadoop目录下include:对外提供的库的头文件lib :对外提供的动态编程库和静态...

  • 回答 4

    HDFS存储机制,包括HDFS的写入过程和读取过程两个部分: 1、写入过程:  1)客户端向namenode请求上传文件,namenode检查目标文件是否已存在,父目录是否存在。2)namenode返回是否可以上传。3)客户端请求第一个 block上传到哪几个datanode服务器上。4)nam...

  • Shuffle 发生在哪里?2021-04-28 20:11
    回答 4

    adoop核心:MapReduce原理。 MR的核心是shuffle,被称为奇迹发生的地方。 shuffle,弄乱,洗牌的意思。partition 分区,sort 排序,spill溢出,disk 磁盘下面是官方对shuffle的配图: phase 阶段,fetch 最终,merge 合并...

  • 回答 2

    Shuffle阶段分为两部分:Map端和Reduce端。一 map端shuffle过程;1-内存预排序:默认每个map有100M内存进行预排序(为了效率),超过阈值,会把内容写到磁盘;    此过程使用快速排序算法;2-根据key和reducer的数量进行分区和排序;首先根据数据所属的Parti...

  • 回答 3

    大数据时代需要1存储大量数据2快速的处理大量数据3从大量数据中进行分析 

  • Hadoop有哪几种模式?2021-04-27 20:20
    回答 3

    hadoop的四种模式。1、本地模式:本地模式就是解压源码包,不需要做任何的配置。通常用于开发调试,或者感受hadoop。2、伪分布模式:在学习当中一般都是使用这种模式,伪分布模式就是在一台机器的多个进程运行多个模块。虽然每一个模块都有相应的进程,但是却...

  • 回答 1

    进入和退出安全模式 [root@localhost bin]# ./hdfs dfsadmin -safemode enter15/08/03 07:26:24 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where ......

没有解决我的问题,去提问