Flink系列 - 實時數(shù)倉之統(tǒng)計數(shù)據(jù)并入redis實戰(zhàn)（七）

??有時候，wordcount 的案例的原理還真是好用，當(dāng)然不過單單是從官網(wǎng)復(fù)制的案例遠(yuǎn)遠(yuǎn)是不滿足我們的平時需求的。那么假如我們?nèi)缦滦枨螅?/p> 

1. 以天為單位，統(tǒng)計各個部門在每小時中銷售的商品數(shù)量，并以日期為組合鍵實時的將結(jié)果放入 redis 中去。   
注意：這個需求有點坑爹，如果我們以普通的滾動和滑動窗口去實現(xiàn)是不會滿足要求的，需求人員說至少1s 計算一次。

數(shù)據(jù)源如下：

{"id":"399","name":"fei niu - 399","sal":283366,"dept":"人事部","ts":1615194501416}
{"id":"398","name":"tang tang - 398","sal":209935,"dept":"燒錢部","ts":1615194501416}
{"id":"395","name":"tang tang - 395","sal":51628,"dept":"帥哥部","ts":1615194501404}
{"id":"400","name":"fei fei - 400","sal":45782,"dept":"燒錢部","ts":1615194501420}
{"id":"401","name":"fei fei - 401","sal":389162,"dept":"帥哥部","ts":1615194501424}
{"id":"402","name":"tang tang - 402","sal":127889,"dept":"人事部","ts":1615194501428}

計算結(jié)果如圖：
image.png
項目結(jié)構(gòu)
image.png
代碼實現(xiàn)
一、創(chuàng)建APP主類，主要的邏輯代碼如下：
public class App {

    private static RedisUtil2 redisUtil2 = RedisUtil2.getInstance();

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

        StreamExecutionEnvironment env = GetStreamExecutionEnvironment.getEnv();
        //請求kafka數(shù)據(jù)
        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers","cdh101:9092");
        prop.setProperty("group.id","cloudera_mirrormaker");
        prop.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
        FlinkKafkaConsumer011<String> myConsumer = new FlinkKafkaConsumer011("luchangyin", new SimpleStringSchema() ,prop);

        //myConsumer.setStartFromGroupOffsets();  // 默認(rèn)行為，從上次消費的偏移量進(jìn)行繼續(xù)消費。
        //myConsumer.setStartFromEarliest(); //Flink從topic中最初的數(shù)據(jù)開始消費
        myConsumer.setStartFromLatest();  //最近的

        //請求kafka數(shù)據(jù)
        DataStreamSource<String> dataStream = env.addSource(myConsumer);

        //dataStream.print();   // {"id":"226","name":"tang tang - 226","sal":280751,"dept":"美女部","ts":1615191802523}

        // ------------ 步驟一：json 解析并統(tǒng)計

       

        // --------------- 步驟二：自定義 redis 的 conditionKey 并將結(jié)算結(jié)果入 redis 中去

        

        // ---------- 步驟三：空實現(xiàn)結(jié)束流程
       

        env.execute("wo xi huan ni");
    }

}
步驟一：json 解析并統(tǒng)計
DataStream<Tuple3<String,String, String>> result = dataStream.map(new MapFunction<String, Employees>() {

            @Override
            public Employees map(String s) throws Exception {
                Employees emp = MyJsonUtils.str2JsonObj(s);
                emp.setEmpStartTime(new Date(emp.getTs()));
                emp.setDt(MyDateUtils.getDate2Hour2(emp.getEmpStartTime()));
                return emp; // Employees(eId=239, eName=tang tang - 239, eSal=286412.0, eDept=人事部, ts=1615191376732, empStartTime=Mon Mar 08 16:16:16 GMT+08:00 2021, dt=2021-03-08 16)
            }
        }).keyBy(new KeySelector<Employees, Tuple2<String,String>>() {
            @Override
            public Tuple2<String, String> getKey(Employees key) throws Exception {
                return new Tuple2<>(key.getDt(),key.getEDept());
            }
        }).window(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8)))
            //.window(TumblingProcessingTimeWindows.of(Time.hours(1)))
            .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(1)))
            .aggregate(new EmpByKeyCountAgg(), new EmpByKeyWindow());

        //result.print(); // (2021-03-08 16,帥哥部,62)

??這里我們自定義了 aggregate 里邊的兩個函數(shù)，用于分類統(tǒng)計結(jié)果值。
EmpByKeyCountAgg 類：

package com.nfdw.function;

import com.nfdw.entity.Employees;
import org.apache.flink.api.common.functions.AggregateFunction;

/** COUNT 統(tǒng)計的聚合函數(shù)實現(xiàn)，每出現(xiàn)一條記錄加一
 *       in, acc, out
 * */
public class EmpByKeyCountAgg implements AggregateFunction<Employees,Long, Long> {

    @Override
    public Long createAccumulator() {
        return 0L;
    }

    @Override
    public Long add(Employees employees, Long aLong) {
        return aLong + 1;
    }

    @Override
    public Long getResult(Long aLong) {
        return aLong;
    }

    @Override
    public Long merge(Long aLong, Long acc1) {
        return aLong + acc1;
    }

}

EmpByKeyWindow 類：

package com.nfdw.function;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import scala.Tuple3;

/** 用于輸出窗口的結(jié)果
 *      in, out, key, window
 * */
public class EmpByKeyWindow implements WindowFunction<Long, Tuple3<String,String, String>, Tuple2<String,String>, TimeWindow> {

    /**
     *     窗口的主鍵，即 itemId
     *     窗口
     *     聚合函數(shù)的結(jié)果，即 count 值
     *     輸出類型為 ItemViewCount
     */
    @Override
    public void apply(Tuple2<String, String> strTuple2, TimeWindow timeWindow, Iterable<Long> iterable, Collector<Tuple3<String, String, String>> collector) throws Exception {
        collector.collect(new Tuple3<String,String, String>(strTuple2.f0,strTuple2.f1, iterable.iterator().next().toString()));
    }

}
步驟二：自定義 redis 的 conditionKey 并將結(jié)算結(jié)果入 redis 中去
DataStream<String> redisSink = result.map(new MapFunction<Tuple3<String, String, String>, String>() {
            @Override
            public String map(Tuple3<String, String, String> str) throws Exception {
                //new Tuple2<String, String>(str.f0.substring(11),str.f2);

                String[] myDate = str._1().split(" ");
                String additionalKey = "index_emp_"+ myDate[0].replaceAll("-","");
                String key = myDate[1];
                double value = Double.valueOf(str._3());
                redisUtil2.zset(additionalKey, key, value);

                return additionalKey +" , "+ key +" , "+ value+ " 成功寫入reids...";
            }
        });

        // redisSink.print(); // index_emp_20210308 , 16 , 54.0 成功寫入reids...

創(chuàng)建操作redis的工具 RedisUtil2 類：

package com.nfdw.utils;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import java.util.Set;

public class RedisUtil2 {
//    private static final Logger log = LoggerFactory.getLogger(RedisUtil.class);

    private static RedisUtil2 instance;

    private static JedisPool jedisPool = RedisPoolUtil2.getPool();

    private RedisUtil2() {
    }

    /**
     * 雙重校驗鎖 保證單例
     *
     * @return
     */
    public static RedisUtil2 getInstance() {
        if (instance == null) {
            synchronized (RedisUtil2.class) {
                if (instance == null) {
                    instance = new RedisUtil2();
                }
            }
        }
        return instance;
    }

    public void zset(String aditionalKey, String key, Double value) {
        Jedis jedis = jedisPool.getResource();
        try {
            jedis.zadd(aditionalKey, value, key);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            closeJedis(jedis);
        }

    }

    public Set<String> myZrange(String aditionalKey) {
        Jedis jedis = jedisPool.getResource();
        Set<String> result = null;
        try {
            result = jedis.zrange(aditionalKey, 0, -1);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            closeJedis(jedis);
        }
        return result;
    }

    /**
     * 通用方法：釋放Jedis
     *
     * @param jedis
     */
    private void closeJedis(Jedis jedis) {
        if (jedis != null) {
            jedis.close();
        }
    }

}
步驟三：空實現(xiàn)結(jié)束流程
redisSink.addSink(new MyAddRedisSink());

解析來我們再實現(xiàn) MyAddRedisSink 類：

package com.nfdw.utils;

import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

public class MyAddRedisSink extends RichSinkFunction<String> {

    @Override
    public void invoke(String value, Context context) throws Exception {
        super.invoke(value, context);
        System.out.println(" sink :"+value);
    }
}

??大致的代碼我們已經(jīng)實現(xiàn)了，當(dāng)然還有事件操作工具類、json實體解析工具類以及 pom文件。
MyJsonUtils 類：

package com.nfdw.utils;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.nfdw.entity.Employees;

public class MyJsonUtils {

    public static Employees str2JsonObj(String str){
        GsonBuilder gsonBuilder = new GsonBuilder();
        gsonBuilder.setPrettyPrinting();
        Gson gson = gsonBuilder.create();
        return gson.fromJson(str, Employees.class);
    }

}

MyDateUtils 類：

package com.nfdw.utils;

import java.text.SimpleDateFormat;
import java.util.Date;

public class MyDateUtils {

//    public static String getDate2Str(){
//        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
//        return sdf.format(new Date());
//    }
//
//    public static long getDate2Timestamp(String ds){
//        //創(chuàng)建SimpleDateFormat對象實例并定義好轉(zhuǎn)換格式
//        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//        Date date = null;
//        try {
//            // 注意格式需要與上面一致，不然會出現(xiàn)異常
//            date = sdf.parse(ds);
//        } catch (ParseException e) {
//            e.printStackTrace();
//        }
//        return date.getTime();
//    }
//
//    public static String getDate2Hour(String ds){
//        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH");
//        Date date = null;
//        String dateStr = null;
//        try {
//            date = sdf.parse(ds);
//            dateStr = df.format(date);
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//        return dateStr;
//    }

    public static String getDate2Hour2(Date date){
        //SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH");
        String dateStr = null;
        try {
            // date = sdf.parse(ds);
            dateStr = df.format(date);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return dateStr;
    }

}

Employees 類：

package com.nfdw.entity;

import com.fasterxml.jackson.annotation.JsonFormat;
import com.google.gson.annotations.SerializedName;
import lombok.Data;
import lombok.experimental.Accessors;
import java.util.Date;

@Data
@Accessors(chain = true)
public class Employees {

    // {"id":"619","name":"fei fei - 619","sal":306875,"dept":"人事部","ts":1615187714251}
    @SerializedName(value = "id")
    private String eId = "";
    @SerializedName(value = "name")
    private String eName = "";
    @SerializedName(value = "sal")
    private double eSal = 0;
    @SerializedName(value = "dept")
    private String eDept = "";
    @SerializedName(value = "ts")
    private long ts = 0L;

    @JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")
    private Date empStartTime;

    private String dt = "";

}

pom.xml 文件如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>MyFirstBigScreen</artifactId>
        <groupId>org.example</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>TestMyCountDemon</artifactId>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <flink.version>1.10.1</flink.version>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.2</version>
            <scope>provided</scope>
        </dependency>

        <!-- redis -->
        <dependency>
            <groupId>org.apache.bahir</groupId>
            <artifactId>flink-connector-redis_2.10</artifactId>
            <version>1.0</version>
        </dependency>
        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>2.9.0</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/com.google.code.gson/gson -->
        <dependency>
            <groupId>com.google.code.gson</groupId>
            <artifactId>gson</artifactId>
            <version>2.8.6</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
            <!--<scope>provided</scope>-->
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.11</artifactId>
            <version>${flink.version}</version>
            <!--<scope>provided</scope>-->
        </dependency>

    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>

            <plugin>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.0.0</version>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.1.1</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <artifactSet>
                                <excludes>
                                    <exclude>com.google.code.findbugs:jsr305</exclude>
                                    <exclude>org.slf4j:*</exclude>
                                    <exclude>log4j:*</exclude>
                                </excludes>
                            </artifactSet>
                            <filters>
                                <filter>
                                    <!-- Do not copy the signatures in the META-INF folder.
                                    Otherwise, this might cause SecurityExceptions when using the JAR. -->
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

??由于我們是自定義的 conditionKey，flink的sink接口還未提供這個功能，因此需要我們自行處理，除了以上實現(xiàn)方式之外，也可以修改源碼進(jìn)行處理，可以參考這篇文章：https://my.oschina.net/u/4596020/blog/4517377
??好了，案例到此為止，直接復(fù)制咱貼就可以用了，希望對你有幫助哦。。。

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容