Flink数据源问题之读取mysql报错如何解决

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: Flink数据源是指Apache Flink用于读取外部系统数据的接口或组件;本合集将探讨Flink数据源的类型、配置方法和最佳实践,以及在使用数据源时可能遇到的错误和解决方案。

问题一:flink sql 1.9 可以通过sql的方式 join 关联外部数据源吗?


想问一下,目前flink sql 1.9 可以通过sql的方式 join 关联外部数据源吗?

比如说通过sql DDL 定义一个Hbase的维表数据源,然后用sql join关联上。

CREATETABLEMyUserTable(

hbase_rowkey_namerowkey_type,

hbase_column_family_name1ROW<...>,

hbase_column_family_name2ROW<...>

)WITH(

'connector.type'='hbase',...

)

SELECT

tb1.xx

MyUserTable.xx

FROM

tb1

JOIN

MyUserTable ON

tb1.xx = MyUserTable.xx*来自志愿者整理的flink邮件归档


参考回答:

正好在看这篇文档,但是我想做一个csv的数据源,不知道格式怎么配置。

https://yq.aliyun.com/articles/717034


关于本问题的更多回答可点击原文查看:https://developer.aliyun.com/ask/373768?spm=a2c6h.14164896.0.0.14f6d9dfoWFkCZ


问题二:Flink sql 支持在流式数据上使用LAG和LEAD函数吗


如何在流式数据源上使用分析函数LAG和EAD函数:

kafka输入数据如:

{"t":"2020-04-01T05:00:00Z", "id":"1", "speed":1.0}

{"t":"2020-04-01T05:05:00Z", "id":"1", "speed":2.0}

{"t":"2020-04-01T05:10:00Z", "id":"1", "speed":3.0}

{"t":"2020-04-01T05:15:00Z", "id":"1", "speed":4.0}

{"t":"2020-04-01T05:20:00Z", "id":"1", "speed":5.0}

{"t":"2020-04-01T05:25:00Z", "id":"1", "speed":6.0}

sql如下:

INSERT INTO topic_sink
SELECT
  t,
  id,
  speed,
  LAG(speed, 1) OVER w AS speed_1,
  LAG(speed, 2) OVER w AS speed_2
FROM topic_source
WINDOW w AS (
      PARTITION BY id
      ORDER BY t
)

我期望得到的结果数据是

{"t":"2020-04-01T05:00:00Z", "id":"1", "speed":1.0, "speed_1":null, "speed_2":null}

{"t":"2020-04-01T05:05:00Z", "id":"1", "speed":2.0,"speed_1":1.0, "speed_2":null}

{"t":"2020-04-01T05:10:00Z", "id":"1", "speed":3.0,"speed_1":2.0, "speed_2":1.0}

{"t":"2020-04-01T05:15:00Z", "id":"1", "speed":4.0,"speed_1":3.0, "speed_2":2.0}

{"t":"2020-04-01T05:20:00Z", "id":"1", "speed":5.0,"speed_1":4.0, "speed_2":3.0}

{"t":"2020-04-01T05:25:00Z", "id":"1", "speed":6.0",speed_1":5.0, "speed_2":4.0}

实际得到的结果数据是:

{"t":"2020-04-01T05:00:00Z", "id":"1", "speed":1.0, "speed_1":1.0, "speed_2":1.0}

{"t":"2020-04-01T05:05:00Z", "id":"1", "speed":2.0,"speed_1":2.0, "speed_2":2.0}

{"t":"2020-04-01T05:10:00Z", "id":"1", "speed":3.0,"speed_1":3.0, "speed_2":3.0}

{"t":"2020-04-01T05:15:00Z", "id":"1", "speed":4.0,"speed_1":4.0, "speed_2":4.0}

{"t":"2020-04-01T05:20:00Z", "id":"1", "speed":5.0,"speed_1":5.0, "speed_2":5.0}

{"t":"2020-04-01T05:25:00Z", "id":"1", "speed":6.0",speed_1":6.0, "speed_2":6.0}

想问一下flink sql里的LAG函数能完成我期望的计算吗?如果可以sql该如何写?


参考回答:

INSERT INTO topic_sink
SELECT
  t,
  id,
  speed,
  LAG(speed, 1, null) OVER w AS speed_1,
  LAG(speed, 2, null) OVER w AS speed_2
FROM topic_source
WINDOW w AS (
      PARTITION BY id
      ORDER BY t
)

这样应该可以


关于本问题的更多回答可点击原文查看:https://developer.aliyun.com/ask/306833?spm=a2c6h.14164896.0.0.6c93d9dfynWHtw


问题三:flink table使用cdc读取mysql数据源报错 unknown error 1227. mysql用户也有replicarion slave 还有replication client权限,请问下这个问题还有可能是什么原因引起的呢?


flink table使用cdc读取mysql数据源报错 unknown error 1227. mysql用户也有replicarion slave 还有replication client权限,请问下这个问题还有可能是什么原因引起的呢?


关于本问题的更多回答可点击原文查看:https://developer.aliyun.com/ask/460625?spm=a2c6h.14164896.0.0.5c71d9dfBYG7Od


问题四:linksql读取tidb全量数据的时候,没有读全,源有1000w+数据,只读了100w+,但是程序也没挂, 看日志,有这么一行,感觉是切分主键的时候乱码了, cdc源表是string类型的主键, , 这种表没读全的,一般是什么原因呢,


请问下, flinksql读取tidb全量数据的时候,没有读全,源有1000w+数据,只读了100w+,但是程序也没挂, 看日志,有这么一行,感觉是切分主键的时候乱码了, cdc源表是string类型的主键, , 这种表没读全的,一般是什么原因呢,



参考回答:

这个可以提一个issue,然后提供sql和样本数据以供复现。


关于本问题的更多回答可点击原文查看:https://developer.aliyun.com/ask/454540?spm=a2c6h.14164896.0.0.6cb9d9dfb0yj4F


问题五:flink数据sum后结果超过源数据


源数据为mysql数据,通过提取mysql数据后写入到kafka,flink接收数据做数据量统计,统计后出现统计量大于原始数据量的情况

CheckpointingMode(CheckpointingMode.EXACTLY_ONCE) 并且在数据统计之前做了去重操作!

请问有人遇到这种问题么?


参考回答:

sum之前先进行groupby去重,flink的上游会多次发生消息会导致重复计算


关于本问题的更多回答可点击原文查看:https://developer.aliyun.com/ask/281738?spm=a2c6h.14164896.0.0.6cb9d9dfb0yj4F

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
目录
相关文章
|
27天前
|
关系型数据库 MySQL API
Flink CDC产品常见问题之mysql整库同步到starrock时任务挂掉如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
27天前
|
API 数据库 流计算
有大佬知道在使用flink cdc实现数据同步,如何实现如果服务停止了对数据源表的某个数据进行删除操作,重启服务之后目标表能进行对源表删除的数据进行删除吗?
【2月更文挑战第27天】有大佬知道在使用flink cdc实现数据同步,如何实现如果服务停止了对数据源表的某个数据进行删除操作,重启服务之后目标表能进行对源表删除的数据进行删除吗?
42 3
|
11天前
|
关系型数据库 MySQL 数据库
运行flyway报错, MySQL 5.6 is no longer supported by Flyway Community Edition
运行flyway报错, MySQL 5.6 is no longer supported by Flyway Community Edition
16 1
|
20天前
|
Java 关系型数据库 MySQL
Flink1.18.1和CDC2.4.1 本地没问题 提交任务到服务器 报错java.lang.NoClassDefFoundError: Could not initialize class io.debezium.connector.mysql.MySqlConnectorConfig
【2月更文挑战第33天】Flink1.18.1和CDC2.4.1 本地没问题 提交任务到服务器 报错java.lang.NoClassDefFoundError: Could not initialize class io.debezium.connector.mysql.MySqlConnectorConfig
32 2
|
27天前
|
关系型数据库 MySQL API
Flink CDC产品常见问题之mysql整库同步到starrock时任务挂掉如何解决
Flink CDC产品常见问题之mysql整库同步到starrock时任务挂掉如何解决
|
27天前
|
分布式计算 Hadoop Java
Flink CDC产品常见问题之tidb cdc 数据量大了就疯狂报空指针如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
27天前
|
资源调度 关系型数据库 测试技术
Flink CDC产品常见问题之没有报错但是一直监听不到数据如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
5天前
|
关系型数据库 MySQL 数据库
mysql卸载、下载、安装(window版本)
mysql卸载、下载、安装(window版本)
|
30天前
|
关系型数据库 MySQL 数据库连接
关于MySQL-ODBC的zip包安装方法
关于MySQL-ODBC的zip包安装方法
|
24天前
|
关系型数据库 MySQL 数据库
rds安装数据库客户端工具
安装阿里云RDS的数据库客户端涉及在本地安装对应类型(如MySQL、PostgreSQL)的客户端工具。对于MySQL,可选择MySQL Command-Line Client或图形化工具如Navicat,安装后输入RDS实例的连接参数进行连接。对于PostgreSQL,可以使用`psql`命令行工具或图形化客户端如PgAdmin。首先从阿里云控制台获取连接信息,然后按照官方文档安装客户端,最后配置客户端连接以确保遵循安全指引。
73 1

相关产品

  • 实时计算 Flink版