连接rabbitmq的时候java有很多包支持，想问一下pyflink有类似的连接支持么？
我是一个flink的初学者。目前的需求是从rabbitmq读入流处理后存进mysql。 因为公司大家的技术栈是python，最近看到了中国社区在推广pyflink（apache-flink）所以就试了一下。但是我发现连接rabbitmq的时候，java有很多包支持，比如 org.apache.flink.streaming.connectors.rabbitmq.* ，想问一下pyflink有类似的连接支持么？
或者有什么推荐的解决方案么比如用pika连接等*来自志愿者整理的FLINK邮件归档
写回答
-
PyFlink目前只支持Python Table API，rabbitmq目前还没有提供Table/SQL的connector，如果想在PyFlink里使用rabbitmq，有以下几种解决方案：
1）实现Java的rabbitmq的TableSource/TableSink，可以参考Kafka等connector的实现，基本只需要在现有实现的基础上包装一下即可。 2）在PyFlink作业里使用rabbitmq的source/sink。目前在PyFlink里注册TableSource/TableSink有2种方式： 2.1）如果使用TableEnvironment.register_table_source/register_table_sink接口注册rabbitmq的connector，你需要做的事情是：写一个Python的rabbitmq的TableSource/TableSink的wrapper，可以参考一下CsvTableSource/CsvTableSource的实现方式[1][2] 2.2) 如果使用TableEnvironment.connect接口，你需要做的事情是：写一个Java的rabbitmq的TableSourceSinkFactory，然后使用CustomConnectorDescriptor[3]，CustomConnectorDescriptor的使用可以参考一下test case[4]:
[1] https://github.com/apache/flink/blob/master/flink-python/pyflink/table/sinks.py#L40 https://github.com/apache/flink/blob/master/flink-python/pyflink/table/sinks.py#L40 [2] https://github.com/apache/flink/blob/master/flink-python/pyflink/table/sources.py#L35 https://github.com/apache/flink/blob/master/flink-python/pyflink/table/sources.py#L35 [3] https://github.com/apache/flink/blob/master/flink-python/pyflink/table/descriptors.py#L1199 https://github.com/apache/flink/blob/master/flink-python/pyflink/table/descriptors.py#L1199 [4] https://github.com/apache/flink/blob/master/flink-python/pyflink/table/tests/test_descriptor.py#L347 https://github.com/apache/flink/blob/master/flink-python/pyflink/table/tests/test_descriptor.py#L347*来自志愿者整理的FLINK邮件归档
2021-12-03 18:57:32赞同 展开评论 打赏
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
