问题一：flink sql消费kafka join普通表为何会性能爬坡？
大家好，
flink sql消费kafka join普通表是会性能爬坡吗？
背景是flink 1.12.0 使用flink sql在yarn per-job发布，消费kafka topic=trades，然后join 数据库里的维表 shop_meta 现在发现每次重启flink sql job，或上游突然增加大量写入时，flink sql的消费速度总是慢慢增加上来，这样就会造成上游积压，等flink sql消费速度上来之后才能慢慢把积压消费完毕。
更多的信息： trades是avro格式，大概有10个字段，但其中有一个字段full_info是一个大json，我这边写了处理json的UDF，就为每个字段都需要处理那个大json。最后生成将近25个字段写下游kafka shop_meta是普通表，没有时间字段，总共有12个字段，30000行左右。整个表数据和索引加起来是16MB；更新频率非常低。现在读jdbc的配置为lookup.cache.max-rows = 20000；lookup.cache.ttl = 2h；scan.fetch-size = 1000 SQL示例如下
SELECT t.shop_id, s.shop_name, ... CAST(json_path_to_str(full_info, '$.response.trade.price', '0.0') AS DOUBLE) price, " CAST(json_path_to_str(full_info, '$.response.trade.payment', '0.0') AS DOUBLE) payment, " CAST(json_path_to_str(full_info, '$.response.trade.total_fee', '0.0') AS DOUBLE) total_fee, " CAST(json_path_to_str(full_info, '$.response.trade.discount_fee', '0.0') AS DOUBLE) discount_fee, " CAST(json_path_to_str(full_info, '$.response.trade.adjust_fee', '0.0') AS DOUBLE) adjust_fee, " CAST(json_path_to_str(full_info, '$.response.trade.received_payment', '0.0') AS DOUBLE) received_payment, " CAST(json_path_to_str(full_info, '$.response.trade.post_fee', '0.0') AS DOUBLE) post_fee, " json_path_to_str(full_info, '$.response.trade.receiver_name', '') receiver_name, " json_path_to_str(full_info, '$.response.trade.receiver_country', '') receiver_country, " json_path_to_str(full_info, '$.response.trade.receiver_state', '') receiver_state, " json_path_to_str(full_info, '$.response.trade.receiver_city', '') receiver_city, " FROM trades t LEFT JOIN shop_meta FOR SYSTEM_TIME AS OF t.proc_time AS s ON t.shop_id=s.shop_id
考虑到整个job里只有简单的ETL，不涉及中间状态，flink对task_manager的配置为 taskmanager.memory.managed.fraction = 0.1 taskmanager.memory.network.fraction = 0.05 实际运行中，task_manager总内存为6G，6 slots，最大并行度为6，所以只有一个task manager。 在监控页面看到task heap=4.13 GB，实际使用heap_used指标比较稳定。 在监控页面中可以看到随着消费速度越来越快，task manager CPU利用率越来越高，KafkaConsumer_topic_partition_currentOffsets - KafkaConsumer_topic_partition_committedOffsets 也在随着消费速度上涨，新生代GC次数和时间也在上涨 当消费完积压后，前两个指标降低，新生代GC趋于平稳
请问有什么调查或解决的方向吗？*来自志愿者整理的flink邮件归档
参考答案：
为了测试到底是因为SQL里的parse json导致性能瓶颈，还是因为join维表 我在SQL中去掉join维表之后重启，发现只需要70s即可达到消费速度=3.8k，整个因为重启导致的积压被压缩到3分钟
所以应该是维表JOIN的问题 现在连的数据库是TiDB，连接串属性为 useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&rewriteBatchedStatements=true来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370040?spm=a2c6h.13066369.question.8.33bf585fyxGC7d
问题二：flink mysql cdc疑问
大佬们，有几个问题想问下，关于flink mysql cdc 1.flink mysql cdc一个任务同步同一个mysql实例的多个表时，会不会创建多个同步管道，读取全部的binlog数据在客户端过滤，可能造成mysql 流量过大呢？*来自志愿者整理的flink邮件归档
参考答案：
其实我不是和别建议直接用CDC，一般用分布式存储的消息去解决，或者你做多个从库，从从库拉取binlog来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370041?spm=a2c6h.13066369.question.7.33bf585fn7Im55
问题三：flink 1.12.2 sql-cli 写入Hive报错 is_generic
当配置好HiveCatalog后， SQL-Cli 也可以查到hive库表信息 创建kafka表：
create table test.test_kafka( word VARCHAR ) WITH ( 'connector' = 'kafka', 'topic' = 'xx', 'scan.startup.mode' = 'latest-offset', 'properties.bootstrap.servers' = 'xx', 'properties.group.id' = 'test', 'format' = 'json', 'json.ignore-parse-errors' = 'true' ); 在 Hive里面可以查到改表 hive > DESCRIBE FORMATTED test_kafka ........... is_generic true .........
但是我在 Flink SQL > select * from test.test_kafka; 报错： org.apache.flink.table.api.ValidationException: Unsupported options found for connector 'kafka'. Unsupported options: is_generic Supported options: connector format json.fail-on-missing-field json.ignore-parse-errors*来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370042?spm=a2c6h.13066369.question.10.33bf585f0c3xAA
问题四：設置look up table source
hi all,
flink在使用temporal join只支持look up table source。我在做單元測試的時候， 下載了hive 表裡面的數據，嘗試了用filesystem註冊temporal table。可是後來發現file system不支持lookup。查詢了文檔( https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/legacySourceSinks.html)，用戶可以自定義look up table source。可是我找不到類似用csv設置look up table source的方法。所以想請問有什麼例子我可以參考嗎？
Regards,*来自志愿者整理的flink邮件归档
参考答案：
你可以参考 JdbcDynamicTableSource [1] 这个 table source 实现了 LookupTableSource 接口，你需要写一个类似 JdbcRowDataLookupFunction 即可 的函数即可。
祝好， Leonard [1] https://github.com/apache/flink/blob/4be9aff3eccb3808df1f10ef7c30480ec11a9cb0/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableSource.java#L70 来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370043?spm=a2c6h.13066369.question.11.33bf585fTOAPCU
问题五：flink1.11版本 -C 指令并未上传udf jar包
执行指令：flink run
-m yarn-cluster
-C file:///xxxx/flink-demo-1.0.jar
xxxxx
在Client端能够构建成功jobgraph，但是在yarn上会报UDF类找不到。我看Classpath中并未上传该JAR包*来自志愿者整理的flink邮件归档
参考答案：
-C 不会上传对应路径下的 jar，最终会被添加到集群的 classpath 中，需要运行的机器对应的路径下要有同样的Jar包才可以。 可以放在私服或者oss的服务，通过 http 的方式加载的 udf jar -C "http://host:port/xxx.jar" 来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370044?spm=a2c6h.13066369.question.12.33bf585fcrDQUh
