Hi,
几点需要考虑一下:
1. 上游(Source端), 你需要用Flink CDC的MySQL Connector;
2. 下游(Sink端),
看你用哪种数据库,如果没有合适的connector,也需要自定义SInk类,继承RichSinkFunction类,
重载open和invoke等几个方法。
Regards,
Leo
在 2023/8/7 17:04, 小昌同学 写道:
各位老师好
,我这边在本地使用通过继承RichSourceFunction类,实现从MySQL中读取数据,但是为啥程序将MySQL中的全部数据获取出来后,程序就自动停止了啊;
以下是
各位老师好:
背景是这样的[flink1.17.1],我在window机器,本地单机调用自定义的pythonUDF,下面是我python代码
err=None
@udtf(input_types=DataTypes.STRING(),
result_types=[DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING(),
DataTypes.STRING()])
def run(data_str):
try:
logger.info("input param is ", data_str)
各位老师好:
[上封邮件未正确加载图片,重新发送]
各位老师好:
背景是这样的[flink1.17.1],我在window机器,本地单机调用自定义的pythonUDF,下面是我python代码
err=None
@udtf(input_types=DataTypes.STRING(),
result_types=[DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING(),
DataTypes.STRING()])
def run(data_str):
try:
logger.info("input para
HI
我简化了我python代码
只要udft方法有外部方法,都会有递归问题, 比如
agan_add_iig(),尽管我的agan_add_iig()实现很简单,flink难道不能外部import自定义方法吗??
def agan_add_iig():
return 2
@udtf(input_types=DataTypes.STRING(),
result_types=[DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING(),
DataTypes.STRING()])
def run(data_str):
try:
退订
退订
请问,flink1.17在使用RichFlatMapFunction进行批计算时,如何在数据结束时将状态写入输出的数据流中?
谢谢