Flink部署问题之hive表没有数据如何解决

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: Apache Flink是由Apache软件基金会开发的开源流处理框架,其核心是用Java和Scala编写的分布式流数据流引擎。本合集提供有关Apache Flink相关技术、使用技巧和最佳实践的资源。

问题一:Flink catalog的几个疑问

Hi Flink社区:

有几个疑问希望社区小伙伴们帮忙解答一下:

1.个人感觉Flink很有必要提供一个官方的catalog,用来支持各种connector,比如:kafka,jdbc,hbase等等connector。不知道社区有没有这个打算,目前没有看到对应的flip

2.社区对hive catalog的定位是什么,后续有可能转正为flink 默认的catalog实现吗?

3.hive catalog是不支持大小写敏感的(字段名都是小写),这个后续会带来哪些问题?想征集下大家的意见避免我们以后踩大坑。

*来自志愿者整理的flink邮件归档



参考答案:

  1. Flink 提供了一套catalog的接口,提提供了几个内置的实现:in-memory catalog, hive catalog,

postgres catalog,

可以根据自己的需求选择。也可以实现自定义的catalog。参考 [1]

  1. hive catalog 主要是对接 hive,方便读取现有的hive catalog的meta信息。当然也可以往hive

catalog写新的meta。

是否会转为默认catalog,据我所知,目前没有。

  1. 一般没什么问题。在和其他区分大小写的db对接的时候,可能有问题。

*来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看:

https://developer.aliyun.com/ask/370308?spm=a2c6h.12873639.article-detail.42.29d04378ApxdqJ



问题二:flink-1.11 ddl kafka-to-hive问题

hive-1.2.1

chk 已经成功了(去chk目录查看了的确有chk数据,kafka也有数据),但是hive表没有数据,我是哪里缺少了什么吗?

String hiveSql = "CREATE TABLE stream_tmp.fs_table (\n" +

" host STRING,\n" +

" url STRING," +

" public_date STRING" +

") partitioned by (public_date string) " +

"stored as PARQUET " +

"TBLPROPERTIES (\n" +

" 'sink.partition-commit.delay'='0 s',\n" +

" 'sink.partition-commit.trigger'='partition-time',\n" +

" 'sink.partition-commit.policy.kind'='metastore,success-file'" +

")";

tableEnv.executeSql(hiveSql);

tableEnv.executeSql("INSERT INTO stream_tmp.fs_table SELECT host, url, DATE_FORMAT(public_date, 'yyyy-MM-dd') FROM stream_tmp.source_table");

*来自志愿者整理的flink邮件归档



参考答案:

rolling 策略配一下? https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/filesystem.html#sink-rolling-policy-rollover-interval

*来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看:

https://developer.aliyun.com/ask/370306?spm=a2c6h.12873639.article-detail.43.29d04378ApxdqJ



问题三:FlinkKafkaConsumer API 维表关联

各位好:

请教一下,用FlinkKafkaConsumer API的话,如何支持SQL的方式,和维表关联。(之前用Kafka API&StreamTableDescriptor API是可以的) "select a.id,b.name from kafka_table a " + "join dim_table FOR SYSTEM_TIME AS OF a.proctime as b on a.id = b.user_id";

*来自志愿者整理的flink邮件归档



参考答案:

```你需要用 DDL 去声明这张 kafka 表[1], 目前不建议使用 KafkaStreamTableDescriptor API。[1]:

https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/kafka.html

*来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看:

https://developer.aliyun.com/ask/370305?spm=a2c6h.12873639.article-detail.44.29d04378ApxdqJ



问题四:Flink sql中可以使用自定义窗口触发器吗

rt

*来自志愿者整理的flink邮件归档



参考答案:

目前是不支持的。不过有个实验性功能可以指定提前输出的策略和迟到处理的策略 [1],可能可以满足你的需求。

[1]:

https://github.com/apache/flink/blob/master/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowEmitStrategy.scala#L175

*来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看:

https://developer.aliyun.com/ask/370304?spm=a2c6h.12873639.article-detail.45.29d04378ApxdqJ



问题五:flink 1.11 cdc: kafka中存了canal-json格式的多张表信息,需要按表解析做

mysql表: CREATE TABLE test ( id int(11) NOT NULL, name varchar(255) NOT NULL, time datetime NOT NULL, status int(11) NOT NULL, PRIMARY KEY (id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8

CREATE TABLE status ( id int(11) NOT NULL, name varchar(255) NOT NULL, PRIMARY KEY (id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8

kafka中数据: // 表test 中insert事件 {"data":[{"id":"1745","name":"jindy1745","time":"2020-07-03 18:04:22","status":"0"}],"database":"ai_audio_lyric_task","es":1594968168000,"id":42,"isDdl":false,"mysqlType":{"id":"int(11)","name":"varchar(255)","time":"datetime","status":"int(11)"},"old":null,"pkNames":["id"],"sql":"","sqlType":{"id":4,"name":12,"time":93,"status":4},"table":"test","ts":1594968168789,"type":"INSERT"}

//表status 中的事件 {"data":[{"id":"10","name":"status"}],"database":"ai_audio_lyric_task","es":1595305259000,"id":589240,"isDdl":false,"mysqlType":{"id":"int(11)","name":"varchar(255)"},"old":null,"pkNames":["id"],"sql":"","sqlType":{"id":4,"name":12},"table":"status","ts":1595305259386,"type":"INSERT"}

如何由于kafka中的json动态的变化的,比如新增一个表,如何能转成应对的RowData, 感觉无法直接用JsonRowDeserializationSchema或CanalJsonDeserializationSchema来做处理。

*来自志愿者整理的flink邮件归档



参考答案:

http://apache-flink.147419.n8.nabble.com/flink-1-10-sql-kafka-format-json-schema-json-object-tt4665.html

这个邮件里提到了类似的问题。

https://issues.apache.org/jira/browse/FLINK-18002 这个issue完成后(1.12),你可以将

“data”,“mysqlType”等格式不确定的字段定义为String类型,

下游通过udf自己再解析对应的json*来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看:

https://developer.aliyun.com/ask/370303?spm=a2c6h.12873639.article-detail.46.29d04378ApxdqJ

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
相关文章
|
19天前
|
SQL 分布式计算 Java
bigdata-17-Hive部署与Hive基础
bigdata-17-Hive部署与Hive基础
34 0
|
1月前
|
消息中间件 关系型数据库 Kafka
flink cdc 数据问题之数据丢失如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
29天前
|
API 数据库 流计算
有大佬知道在使用flink cdc实现数据同步,如何实现如果服务停止了对数据源表的某个数据进行删除操作,重启服务之后目标表能进行对源表删除的数据进行删除吗?
【2月更文挑战第27天】有大佬知道在使用flink cdc实现数据同步,如何实现如果服务停止了对数据源表的某个数据进行删除操作,重启服务之后目标表能进行对源表删除的数据进行删除吗?
46 3
|
30天前
|
Oracle 关系型数据库 MySQL
Flink CDC产品常见问题之flink Oraclecdc 捕获19C数据时报错错如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
19天前
|
Kubernetes 流计算 Perl
在Rancher K8s上部署Flink时,TaskManager连接不上并不断重启可能是由多种原因导致的
在Rancher K8s上部署Flink时,TaskManager连接不上并不断重启可能是由多种原因导致的
29 7
|
30天前
|
分布式计算 Hadoop Java
Flink CDC产品常见问题之tidb cdc 数据量大了就疯狂报空指针如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
30天前
|
资源调度 关系型数据库 测试技术
Flink CDC产品常见问题之没有报错但是一直监听不到数据如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
1月前
|
消息中间件 关系型数据库 MySQL
Flink CDC产品常见问题之把flink cdc同步的数据写入到目标服务器失败如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
2月前
|
消息中间件 Kafka Apache
Apache Flink 是一个开源的分布式流处理框架
Apache Flink 是一个开源的分布式流处理框架
476 5
|
29天前
|
SQL Java API
官宣|Apache Flink 1.19 发布公告
Apache Flink PMC(项目管理委员)很高兴地宣布发布 Apache Flink 1.19.0。
1059 1
官宣|Apache Flink 1.19 发布公告

相关产品

  • 实时计算 Flink版