Cloudera流分析中如何引入FlinkSQL

发布时间:2021-12-21 09:28:14 作者:柒染
来源:亿速云 阅读:150

这篇文章将为大家详细讲解有关Cloudera流分析中如何引入FlinkSQL,文章内容质量较高,因此小编分享给大家做个参考,希望大家阅读完这篇文章后对相关知识有一定的了解。

由Apache Flink提供支持的Cloudera Streaming Analytics的1.2.0.0版本提供了广泛的  新功能  ,包括通过  Apache Atlas  支持血缘和元数据跟踪,支持连接到  Apache Kudu  以及期待已久的  FlinkSQL   API 的第一次迭代。  
Flink的SQL接口使流处理民主化,因为它可以迎合比当前广泛使用的Java和Scala API(面向数据工程人群)更大的社区。将SQL推广到流处理和流分析用例提出了一系列挑战:我们必须解决表达无限流和记录的及时性的问题。让我们考虑以下查询:
SELECT  userId,  
 COUNT(*) AS count,
 
 SESSION_START(clicktime,
 INTERVAL '30' MINUTE)
FROM clicksGROUP BY  
 SESSION(clicktime, INTERVAL '30' MINUTE)

 userId
该查询产生每个用户会话的点击计数,该计数由会话之间30分钟的不活动状态定义,并在遇到新会话时实时更新。这是在流处理中已经很好建立的概念的示例,在这种情况下,会话窗口被引入到SQL语法中以表示记录的及时性。重要的是要强调Flink支持的语法是ANSI SQL,它不是特定的方言。实际上,Flink社区正在与Apache Beam和Apache Calcite社区合作,以  统一的方式 应对FlinkSQL的挑战。    
 
转换到流处理组织
从上述查询来看,很明显,更大的用户群可以有效地制定查询,从而为企业增加价值。但是,它给组织带来了以下问题:  

1) 在流媒体领域中可以用SQL制定多少业务逻辑?

2) 这如何改变从开发到生产的流式作业旅程?

3) 这如何影响数据工程团队的范围?

我们认为,今天编写的大多数流查询都可以通过FlinkSQL表示,以提供有根据的猜测,我们希望它能达到今天遇到的流查询的80%左右,这很适合通过此SQL实现API。首先,这似乎有些夸大其词,我们将在下一部分中详细介绍。
当前,我们经常遇到使用Flink的组织,其中近实时获得业务价值是数据工程师的特权。数据分析人员通常是特定领域知识的专家,他们倾向于使用标准MPP或OLAP系统中存储的这些流的快照,例如通过Apache Impala查询存储在Kudu中的数据。这从本质上引入了寻找以流的方式对其进行洞察和生产化。分析师在证实其假设之后,必须与几个数据工程师确保数周甚至数月的项目资金投入,以细致地重新实现已经用另一种语言(通常是SQL)制定的业务逻辑。FlinkSQL使分析人员可以直接与流进行交互,并单击按钮即可部署流作业。
反过来,这又解放了数据工程师,使他们可以专注于具有挑战性的20%的查询,并建立可重用的特定领域的库,这些库可以直接从SQL中作为一组用户自定义函数加以利用。 

 
FlinkSQL的功能
为了展示FlinkSQL的功能,我们最近在我们的标准  教程套件 下发布了  SQL   教程 。让我们在这里重点介绍一些功能。  
小编针对Apache Kafka主题进行操作,其中包含JSON格式的事务条目。让我们为此定义一个表Schema,并指定我们要测量timestamp列记录的时间的流逝(称为  event-time语义 )。
CREATE TABLE ItemTransactions (transactionId    
BIGINT,`timestamp`    
BIGINT,itemId    
STRING,
quantity INT,
event_time AS CAST(from_unixtime(floor(`timestamp`/1000)) 
AS TIMESTAMP(3)),WATERMARK FOR event_time 
AS event_time - INTERVAL '5' SECOND) WITH ('connector.type'      = 'kafka','connector.version'   = 'universal','connector.topic'     = 'transaction.log.1','connector.startup-mode' = 'earliest-offset','
connector.properties.bootstrap.servers' = '<broker_address>','format.type' = 'json');
请注意,在使用事件时间语义时,我们必须指定水印以为Flink提供启发式方法以测量事件时间的经过。这可以是返回时间戳的任意表达式。在较高级别上,水印指定了正确性(无限期等待潜在的延迟到达)和延迟(尽可能快地产生结果)之间的折衷。
创建上表后,我们可以提交以下查询:


SELECT * FROM ItemTransactions LIMIT 10;SELECT TUMBLE_START(event_time, INTERVAL '10' SECOND) as window_start, itemId, sum(quantity) as volumeFROM ItemTransactionsGROUP BY itemId, TUMBLE(event_time, INTERVAL '10' SECOND);
第一个查询提供了直接的采样。使用limit子句是可选的,省略会导致结果以流方式不断更新。第二个查询实现一个简单的窗口聚合。这些查询的结果可以返回到交互式Flink SQL cli,或者可以通过INSERT INTO语句直接写入输出表。
FlinkSQL还提供了更复杂的子句,例如,可以按以下公式来查找在每10分钟的窗口中交易次数最多的前3个项目:


SELECT * FROM ( SELECT * , ROW_NUMBER() OVER (   PARTITION BY window_start   ORDER BY num_transactions desc ) AS rownum FROM (   SELECT TUMBLE_START(event_time, INTERVAL '10' MINUTE) AS window_start, itemId, COUNT(*) AS num_transactions   FROM ItemTransactions   GROUP BY itemId, TUMBLE(event_time, INTERVAL '10' MINUTE) ))WHERE rownum <=3;
除了这些内置语言元素之外,您还可以将以Java和Scala实现的  功能 注册到FlinkSQL环境。 
FlinkSQL还支持访问外部目录以访问存储在外部系统中的Schema和数据,当前,我们支持Hive,Kudu和Schema Registry目录。

 
后续步骤
在当前版本中,提交SQL查询的两个选项是使用SQL CLI或将它们包装到Java程序中。正如我们在最近的  主题演讲中 所讨论的,我们正在积极开发图形用户界面,以帮助进行交互式查询编辑。  
Cloudera流分析中如何引入FlinkSQL
在添加GUI之后,我们将在短期内公开其针对第三方工具的编程后端,以公开与JDBC for FlinkSQL等效的接口,该接口可能更多地基于REST和Kafka构建。

关于Cloudera流分析中如何引入FlinkSQL就分享到这里了,希望以上内容可以对大家有一定的帮助,可以学到更多知识。如果觉得文章不错,可以把它分享出去让更多的人看到。

推荐阅读:
  1. Cloudera实现Hadoop过程中Hadoop设置的示例分析
  2. 如何使用FlinkSQL内置函数

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

flinksql cloudera

上一篇:怎么使用Component的padding功能

下一篇:Component React应用怎么在Component之间跳转

相关阅读

您好,登录后才能下订单哦!

密码登录
登录注册
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》