【pyflink】pyflink扩展jar包开辟指南（Sink篇）

一身转战三千里

1. 媒介

民圆关于pyflink的介绍太少了。
近来有效pyflink处置数据，终极需求sink到redis中，研讨有果，分享于人，以此为例，举一反三。
简朴介绍一下怎样为pyflink开辟jar包。
2. 阐发

我们的需求是正在python中可以参加sink，正在python中如许挪用。

1. data_stream.add_sink(RedisSink("localhost", 6379, "result"))

复造代码

我们察看到add_sink吸取的参数是一个SinkFunction，那是一个java类，以是我们要用java去完成那个类，然后再正在python中引进，给到add_sink办法挪用。
正在java版本的Flink中，有一个曾经用RichSinkFunction完成的RedisSink，不言而喻RichSinkFunction是SinkFunction的子类，构建java没有是本文的重面，感爱好的能够参考以下那个包的源代码。

1. package org.apache.flink.streaming.connectors.redis;

复造代码

java实践要做的是本身的完美，并返回python需求的类。
3. java代码

1. package simple_redis_sink;
3. import org.apache.flink.api.java.tuple.Tuple2;
4. import org.apache.flink.streaming.connectors.redis.RedisSink;
5. import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
6. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
7. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
8. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
10. public class SimpleRedisSink {
11.     public RedisSink redisSink;
13.     // 处置数据，我那里是用的redis的hset，以是需求name, key, value三个参数，此中name我是看成种别（表名）去用的。
14.     public static final class RedisSinkExample implements RedisMapper<Tuple2<String, String>> {
15.         public String name;
16.         
17.         public RedisSinkExample(String name) {
18.             this.name = name;
19.         }
20.         
21.         # SinkRedis初初化时挪用，getCommandDescription，设置插进redis的方法。
22.         public RedisCommandDescription getCommandDescription() {
23.             return new RedisCommandDescription(RedisCommand.HSET, this.name);
24.         }
25.         
26.         // data是一个(key, value)的Tuple，invoke办法中掏出key
27.         public String getKeyFromData(Tuple2<String, String> data) {
28.             return data.f0;
29.         }
31.         // data是一个(key, value)的Tuple，invoke办法中掏出value
32.         public String getValueFromData(Tuple2<String, String> data) {
33.             return data.f1;
34.         }
35.     }
37.     // 初初化，那里将会是python挪用的进口，host、port、name是python传过去的
38.     public SimpleRedisSink(String host, int port, String name) {
39.         FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost(host).setPort(port).build();
40.         this.redisSink = new RedisSink(conf, new SimpleRedisSink.RedisSinkExample(name));
41.     }
43.     // 返回RedisSink类
44.     public RedisSink get() {
45.         return this.redisSink;
46.     }
48. }

复造代码

4. 编译天生jar

• Project structure->Artifacts->+->JAR->From modules with dependencies...
  
• 正在弹出的Create JAR from Modules对话框中，挑选extact to the target JAR一项。
  
• 正在output layout中面击根部，查抄一下Manifest File战Class Path能否设置准确。
  
  
  
  
  
  
  

Artifacts设置完成后，正在主菜单挑选Build->Build Artifacts...，弹出对话框当选择Build选项便可。

5. pyflink中利用jar包

起首需求正在正在env中减载jar包。

1. env = StreamExecutionEnvironment.get_execution_environment()
2. env.add_jars("file:///Users/bitekong/PycharmProjects/jyyc_dp_stream/simple_redis_sink.jar")

复造代码

正在python中编写py4j类，买通python取java的办法挪用。

1. class RedisSink(SinkFunction):
2.     def __init__(self, host, port, name):
3.         gateway = get_gateway()
4.         j_redis_sink = gateway.jvm.simple_redis_sink.SimpleRedisSink(host, port, name).get()
5.         super(RedisSink, self).__init__(sink_func=j_redis_sink)

复造代码

接下去就能够正在data_stream中将方才写好的RedisSink类参加sink啦：

1. data_stream.add_sink(RedisSink("localhost", 6379, "result"))

复造代码

免责声明：假如进犯了您的权益，请联络站少，我们会实时删除侵权内乱容，感谢协作！