flink ETL数据处理

　　　　　　　　　　　　　　　　　　　　　　Flink ETL 实现数据清洗

一：需求(针对算法产生的日志数据进行清洗拆分)

　　1. 算法产生的日志数据是嵌套json格式，需要拆分

　　2.针对算法中的国家字段进行大区转换

　　3.最后把不同类型的日志数据分别进行储存

二：整体架构

　　　　　　这里演示处理从rabbitmq来的数据进行数据处理然后发送到rabbitmq

自定义redistSource flink没有redis的source

package com.yw.source;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import redis.clients.jedis.Jedis;

import redis.clients.jedis.exceptions.JedisConnectionException;

import java.util.HashMap;

import java.util.Map;

/**

 * redis中进行数据初始化

 * <p>

 * 在reids中保存国家和大区关系

 * hset areas AREA_IN IN

 * hset areas AREA_US US

 * hset areas AREA_CT TW,HK

 * hset areas AREA_AR PK,KW,SA

 *

 *

 * @Auther: YW

 * @Date: 2019/6/15 10:23

 * @Description:

 */

public class MyRedisSource implements SourceFunction<HashMap<String, String>> {

    private final Logger LOG = LoggerFactory.getLogger(MyRedisSource.class);

    private boolean isRuning = true;

    private Jedis jedis = null;

    private final long SLEEP = 60000;

    private final long expire = 60;

    @Override

    public void run(SourceContext<HashMap<String, String>> ctx) throws Exception {

        this.jedis = new Jedis("localhost", 6397);

        // 存储国家和地区关系

        HashMap<String, String> map = new HashMap<>();

        while (isRuning) {

            try {

                map.clear(); // 老数据清除

                Map<String, String> areas = jedis.hgetAll("areas");

                for (Map.Entry<String, String> entry : areas.entrySet()) {

                    String key = entry.getKey();

                    String value = entry.getValue();

                    String[] splits = value.split(",");

                    for (String split : splits) {

                        map.put(split, key);

                    }

                }

                if (map.size() > 0) {

                    // map >0 数据发送出去

                    ctx.collect(map);

                }else {

                    LOG.warn("获取数据为空！");

                }

                // 歇6秒

                Thread.sleep(SLEEP);

            } catch (JedisConnectionException e) {

                LOG.error("redis连接异常 重新连接",e.getCause());

                // 如果连接异常 重新连接

                jedis = new Jedis("localhost", 6397);

            }catch (Exception e){

                LOG.error("redis Source其他异常",e.getCause());

            }

        }

    }

    @Override

    public void cancel() {

        isRuning = false;

        while (jedis != null) {

            jedis.close();

        }

    }

}

DataClean数据处理

package com.yw;

import com.alibaba.fastjson.JSONArray;

import com.alibaba.fastjson.JSONObject;

import com.rabbitmq.client.AMQP;

import com.yw.source.MyRedisSource;

import org.apache.flink.streaming.api.CheckpointingMode;

import org.apache.flink.streaming.api.datastream.DataStream;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.environment.CheckpointConfig;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;

import org.apache.flink.streaming.connectors.rabbitmq.RMQSink;

import org.apache.flink.streaming.connectors.rabbitmq.RMQSinkPublishOptions;

import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;

import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;

import org.apache.flink.streaming.util.serialization.SimpleStringSchema;

import org.apache.flink.util.Collector;

import java.util.HashMap;

/**

 * @Auther: YW

 * @Date: 2019/6/15 10:09

 * @Description:

 */

public class DataClean {

    // 队列名

    public final static String QUEUE_NAME = "two.aa.in";

    public static void main(String[] args) throws Exception {

        // 获取环境

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 一分钟 checkpoint

        env.enableCheckpointing(60000);

        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000); // enableCheckpointing最小间隔时间（一半）

        env.getCheckpointConfig().setCheckpointTimeout(10000);// 超时时间

        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        final RMQConnectionConfig rmqConf = new RMQConnectionConfig.Builder().setHost("127.0.0.1").setPort(5672).setVirtualHost("/").setUserName("guest").setPassword("guest").build();

        // 获取mq数据

        DataStream<String> data1 = env.addSource(new RMQSource<String>(rmqConf, QUEUE_NAME, false, new SimpleStringSchema())).setParallelism(1);

        //{"dt":"2019-06-10","countryCode":"US","data":[{"type":"s1","score":0.3,"level":"A"},{"type":"s2","score":0.1,"level":"B"},{"type":"s3","score":0.2,"level":"C"}]}

        DataStreamSource<HashMap<String, String>> mapData = env.addSource(new MyRedisSource());

        // connect可以连接两个流

        DataStream<String> streamOperator = data1.connect(mapData).flatMap(new CoFlatMapFunction<String, HashMap<String, String>, String>() {

            // 保存 redis返回数据  国家和大区的映射关系

            private HashMap<String, String> allMap = new HashMap<String, String>();

            // flatMap1 处理rabbitmq的数据

            @Override

            public void flatMap1(String value, Collector<String> out) throws Exception {

                JSONObject jsonObject = JSONObject.parseObject(value);

                String countryCode = jsonObject.getString("countryCode");

                String dt = jsonObject.getString("dt");

                // 获取大区

                String area = allMap.get(countryCode);

                JSONArray jsonArray = jsonObject.getJSONArray("data");

                for (int i = 0; i < jsonArray.size(); i++) {

                    JSONObject jsonObject1 = jsonArray.getJSONObject(i);

                    jsonObject1.put("area", area);

                    jsonObject1.put("dt", dt);

                    out.collect(jsonObject1.toJSONString());

                }

            }

            // 处理redis的返回的map类型的数据

            @Override

            public void flatMap2(HashMap<String, String> value, Collector<String> out) throws Exception {

                this.allMap = value;

            }

        });

        streamOperator.addSink(new RMQSink<String>(rmqConf, new SimpleStringSchema(), new RMQSinkPublishOptions<String>() {

            @Override

            public String computeRoutingKey(String s) {

                return "CC";

            }

            @Override

            public AMQP.BasicProperties computeProperties(String s) {

                return null;

            }

            @Override

            public String computeExchange(String s) {

                return "test.flink.output";

            }

        }));

        data1.print();

        env.execute("etl");

    }

}

rabbitmq 模拟数据

package com.yw;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

import java.text.SimpleDateFormat;

import java.util.Date;

import java.util.Random;

/**

 * @Auther: YW

 * @Date: 2019/6/5 14:57

 * @Description:

 */

public class RabbitMQProducerUtil {

    public final static String QUEUE_NAME = "two.aa.in";

    public static void main(String[] args) throws Exception {

        //创建连接工厂

        ConnectionFactory factory = new ConnectionFactory();

        //设置RabbitMQ相关信息

        factory.setHost("127.0.0.1");

        factory.setUsername("guest");

        factory.setPassword("guest");

        factory.setVirtualHost("/");

        factory.setPort(5672);

        //创建一个新的连接

        Connection connection = factory.newConnection();

        //创建一个通道

        Channel channel = connection.createChannel();

        // 声明一个队列

//        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        //发送消息到队列中

        String message = "{\"dt\":\""+getCurrentTime()+"\",\"countryCode\":\""+getCountryCode()+"\"," +

                "{\"type\":\""+getType()+"\",\"score\":"+getScore()+"\"level\":\""+getLevel()+"\"}," +

                "{\"type\":\""+getType()+"\",\"score\":"+getScore()+"\"level\":\""+getLevel()+"\"}," +

                "{\"type\":\""+getType()+"\",\"score\":"+getScore()+"\"level\":\""+getLevel()+"\"}]}";

        //我们这里演示发送一千条数据

        for (int i = 0; i < 20; i++) {

                channel.basicPublish("", QUEUE_NAME, null, (message + i).getBytes("UTF-8"));

              System.out.println("Producer Send +'" + message);

        }

        //关闭通道和连接

        channel.close();

        connection.close();

    }

    public static String getCurrentTime() {

        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        return sdf.format(new Date());

    }

    public static String getCountryCode() {

        String[] types={"US","TN","HK","PK","KW","SA","IN"};

        Random random = new Random();

        int i = random.nextInt(types.length);

        return types[i];

    }

    public static String getType() {

        String[] types={"s1","s2","s3","s4","s5"};

        Random random = new Random();

        int i = random.nextInt(types.length);

        return types[i];

    }

    public static String getScore() {

        String[] types={"0.1","0.2","0.3","0.4","0.5"};

        Random random = new Random();

        int i = random.nextInt(types.length);

        return types[i];

    }

    public static String getLevel() {

        String[] types={"A","B","C","D","E"};

        Random random = new Random();

        int i = random.nextInt(types.length);

        return types[i];

    }

}

redis 初始化数据

* hset areas AREA_IN IN
* hset areas AREA_US US
* hset areas AREA_CT TW,HK
* hset areas AREA_AR PK,KW,SA

------------最后运行DataClean------------

巴特西

flink ETL数据处理

最新文章

热门文章