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

问题一：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