# Flink Join样例程序（Java） #### 功能介绍 在Flink应用中，调用flink-connector-kafka模块的接口，生产并消费数据。 #### 代码样例 用户在开发前需要使用对接安全模式的Kafka，则需要引入kafka-clients-\*.jar，该jar包可在client目录下获取。 下面列出producer和consumer，以及Flink Stream SQL Join使用主要逻辑代码作为演示。 完整代码参见com.huawei.bigdata.flink.examples.WriteIntoKafka和com.huawei.bigdata.flink.examples.SqlJoinWithSocket 1. 每秒钟往Kafka中生产一条用户信息，用户信息由姓名、年龄、性别组成。 ``` //producer代码 public class WriteIntoKafka { public static void main(String[] args) throws Exception { // 打印出执行flink run的参考命令 System.out.println("use command as: "); System.out.println("./bin/flink run --class com.huawei.bigdata.flink.examples.WriteIntoKafka" + " /opt/test.jar --topic topic-test -bootstrap.servers 10.91.8.218:9092"); System.out.println("./bin/flink run --class com.huawei.bigdata.flink.examples.WriteIntoKafka" + " /opt/test.jar --topic topic-test -bootstrap.servers 10.91.8.218:21007 --security.protocol SASL_PLAINTEXT --sasl.kerberos.service.name kafka"); System.out.println("******************************************************************************************"); System.out.println(" is the kafka topic name"); System.out.println(" is the ip:port list of brokers"); System.out.println("******************************************************************************************"); // 构造执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 设置并发度 env.setParallelism(1); // 解析运行参数 ParameterTool paraTool = ParameterTool.fromArgs(args); // 构造流图，将自定义Source生成的数据写入Kafka DataStream messageStream = env.addSource(new SimpleStringGenerator()); FlinkKafkaProducer producer = new FlinkKafkaProducer<>(paraTool.get("topic"), new SimpleStringSchema(), paraTool.getProperties()); producer.setWriteTimestampToKafka(true); messageStream.addSink(producer); // 调用execute触发执行 env.execute(); } // 自定义Source，每隔1s持续产生消息 public static class SimpleStringGenerator implements SourceFunction { static final String[] NAME = {"Carry", "Alen", "Mike", "Ian", "John", "Kobe", "James"}; static final String[] SEX = {"MALE", "FEMALE"}; static final int COUNT = NAME.length; boolean running = true; Random rand = new Random(47); @Override //rand随机产生名字，性别，年龄的组合信息 public void run(SourceContext ctx) throws Exception { while (running) { int i = rand.nextInt(COUNT); int age = rand.nextInt(70); String sexy = SEX[rand.nextInt(2)]; ctx.collect(NAME[i] + "," + age + "," + sexy); thread.sleep(1000); } } @Override public void cancel() { running = false; } } } ``` 2. 生成Table1和Table2，并使用Join对Table1和Table2进行联合查询，打印输出结果。 ``` public class SqlJoinWithSocket { public static void main(String[] args) throws Exception{ final String hostname; final int port; System.out.println("use command as: "); System.out.println("flink run --class com.huawei.bigdata.flink.examples.SqlJoinWithSocket" + " /opt/test.jar --topic topic-test -bootstrap.servers xxxx.xxx.xxx.xxx:9092 --hostname xxx.xxx.xxx.xxx --port xxx"); System.out.println("flink run --class com.huawei.bigdata.flink.examples.SqlJoinWithSocket" + " /opt/test.jar --topic topic-test -bootstrap.servers xxxx.xxx.xxx.xxx:21007 --security.protocol SASL_PLAINTEXT --sasl.kerberos.service.name kafka" + "--hostname xxx.xxx.xxx.xxx --port xxx"); System.out.println("******************************************************************************************"); System.out.println(" is the kafka topic name"); System.out.println(" is the ip:port list of brokers"); System.out.println("******************************************************************************************"); try { final ParameterTool params = ParameterTool.fromArgs(args); hostname = params.has("hostname") ? params.get("hostname") : "localhost"; port = params.getInt("port"); } catch (Exception e) { System.err.println("No port specified. Please run 'FlinkStreamSqlJoinExample " + "--hostname --port ', where hostname (localhost by default) " + "and port is the address of the text server"); System.err.println("To start a simple text server, run 'netcat -l -p ' and " + "type the input text into the command line"); return; } EnvironmentSettings fsSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build(); StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, fsSettings); //基于EventTime进行处理 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); env.setParallelism(1); ParameterTool paraTool = ParameterTool.fromArgs(args); //Stream1，从Kafka中读取数据 DataStream> kafkaStream = env.addSource(new FlinkKafkaConsumer<>(paraTool.get("topic"), new SimpleStringSchema(), paraTool.getProperties())).map(new MapFunction>() { @Override public Tuple3 map(String s) throws Exception { String[] word = s.split(","); return new Tuple3<>(word[0], word[1], word[2]); } }); //将Stream1注册为Table1 tableEnv.registerDataStream("Table1", kafkaStream, "name, age, sexy, proctime.proctime"); //Stream2，从Socket中读取数据 DataStream> socketStream = env.socketTextStream(hostname, port, "\n"). map(new MapFunction>() { @Override public Tuple2 map(String s) throws Exception { String[] words = s.split("\\s"); if (words.length < 2) { return new Tuple2<>(); } return new Tuple2<>(words[0], words[1]); } }); //将Stream2注册为Table2 tableEnv.registerDataStream("Table2", socketStream, "name, job, proctime.proctime"); //执行SQL Join进行联合查询 Table result = tableEnv.sqlQuery("SELECT t1.name, t1.age, t1.sexy, t2.job, t2.proctime as shiptime\n" + "FROM Table1 AS t1\n" + "JOIN Table2 AS t2\n" + "ON t1.name = t2.name\n" + "AND t1.proctime BETWEEN t2.proctime - INTERVAL '1' SECOND AND t2.proctime + INTERVAL '1' SECOND"); //将查询结果转换为Stream，并打印输出 tableEnv.toAppendStream(result, Row.class).print(); env.execute(); } } ```