date:20230810

Re: Flink消费MySQL

2023-08-10 Thread Leo

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中java调用pythonUDF奇怪问题[maximum recursion depth][python函数正常]

2023-08-10 Thread 1

各位老师好: 背景是这样的[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中java调用pythonUDF奇怪问题[maximum recursion depth][python函数正常]

2023-08-10 Thread 1

各位老师好: [上封邮件未正确加载图片,重新发送] 各位老师好: 背景是这样的[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

在flink1.17.1中java调用pythonUDF奇怪问题[maximum recursion depth][python函数正常]

2023-08-10 Thread 1

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: