Flink实时计算运用（六）Flink 连接器的使用

通过Sink写入HDFS数据

功能： 将Socket接收到的数据， 写入至HDFS文件中。

代码：HdfsSinkApplication实现类

public class HdfsSinkApplication { 
          

    public static void main(String[] args) throws Exception{ 
          
        // 配置日志文件
        System.setProperty("log4j.configurationFile", "log4j2.xml");

        // 1. 创建运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 读取Socket数据源
        DataStreamSource<String> socketStr = env.socketTextStream("localhost", 9911, "\n");
        BucketingSink<String> sink = new BucketingSink<String>("d:/tmp/hdfs");
        sink.setBucketer(new DateTimeBucketer<>("yyyy-MM-dd--HHmm"));
        sink.setWriter(new StringWriter())
            .setBatchSize(5*1024) // 设置每个文件的大小
            .setBatchRolloverInterval(5*1000) // 设置滚动写入新文件的时间
            .setInactiveBucketCheckInterval(30*1000) // 30秒检查一次不写入的文件
            .setInactiveBucketThreshold(60*1000); // 60秒不写入，就滚动写入新的文件
        socketStr.addSink(sink).setParallelism(1);

        // 5. 执行任务
        env.execute("job");

    }

}

POM依赖:

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-filesystem_2.11</artifactId>
    <version>1.11.2</version>
</dependency>

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>2.8.1</version>
</dependency>

数据源模拟实现， SocketSourceApplication实现类：

public class SocketSourceApplication { 
              
    /** * 服务端的端口 */
    private int port;

    /** * 初始化构造方法 * @param port */
    public SocketSourceApplication(int port) { 
          
        this.port = port;
    }
  /** * IP 访问列表 */
    private static String[] accessIps = new String[]{ 
          "10.10.20.101", "10.10.20.102", "10.10.20.103"};

    /** * 请求访问类型 */
    private static String[] accessTypes = new String[] { 
          "GET", "POST", "PUT"};

    /** * 请求接口信息 */
    private static String[] accessApis = new String[] { 
          "addOrder", "getAccount", "getOrder"};    

    /** * Netty通讯服务启动方法 * @throws Exception */
    public void runServer() throws Exception  { 
          
        // 1. 创建Netty服务
        // 2. 定义事件Boss监听组
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        // 3. 定义用来处理已经被接收的连接
        EventLoopGroup workerGourp = new NioEventLoopGroup();
        try { 
          
            // 4. 定义NIO的服务启动类
            ServerBootstrap sbs = new ServerBootstrap();
            // 5. 配置NIO服务启动的相关参数
            sbs.group(bossGroup, workerGourp)
                    .channel(NioServerSocketChannel.class)
                    // tcp最大缓存链接个数,它是tcp的参数, tcp_max_syn_backlog(半连接上限数量, CENTOS6.5默认是128)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    //保持连接的正常状态
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    // 根据日志级别打印输出
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() { 
          
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception { 
          
                            //管道注册handler
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            //编码通道处理
                            pipeline.addLast("decode", new StringDecoder());
                            //转码通道处理
                            pipeline.addLast("encode", new StringEncoder());
                            // 处理接收到的请求
                            pipeline.addLast(new NettyServerHandler());
                        }
                    });

            System.err.println("-------server 启动------");
            // 6. 监听控制台的输入, 并将输入信息, 广播发送给客户端
            new Thread(new Runnable() { 
          
                @Override
                public void run() { 
          
                    try { 
          
                        while(true) { 
          
                            String accessLog = getAccessLog();
                            System.out.println("broadcast (" + NettyServerHandler.channelList.size() + ") ==> " + accessLog);
                            if(NettyServerHandler.channelList.size()  > 0 ){ 
          
                                for(Channel channel : NettyServerHandler.channelList) { 
          
                                    channel.writeAndFlush(accessLog);
                                }
                            }
                            Thread.sleep(1000);
                        }
                    }catch(Exception e) { 
          
                        e.printStackTrace();
                    }
                }
            }).start();

            // 7. 启动netty服务
            ChannelFuture cf = sbs.bind(port).sync();
            cf.channel().closeFuture().sync();

        }catch (Exception e) { 
          
            e.printStackTrace();
        }
    }

    /** * 获取访问日志 * @return */
    private String getAccessLog() { 
          
        StringBuilder strBuilder = new StringBuilder();
        strBuilder.append(accessIps[new Random().nextInt(accessIps.length )]).append("\t")
                .append(System.currentTimeMillis()).append("\t")
                .append(accessTypes[new Random().nextInt(accessTypes.length)]).append("\t")
                .append(accessApis[new Random().nextInt(accessApis.length)]).append("\t\n");
        return strBuilder.toString();
    }

    /** * netty服务端的启动 * @param args * @throws Exception */
    public static void main(String[] args) throws Exception{ 
          
        new SocketSourceApplication(9911).runServer();
    }

}

NettyServerHandler实现类: