3.Flink实时项目之流程分析及环境搭建

时间：2022-12-30 08:40:14

1. 流程分析

前面已经将日志数据（ods_base_log）及业务数据（ods_base_db_m）发送到kafka，作为ods层，接下来要做的就是通过flink消费kafka 的ods数据，进行简单的处理作为dwd层，然后再写回到kafka。

3.Flink实时项目之流程分析及环境搭建

每层职能

分层 数据描述 计算工具 存储介质
ODS 原始数据，日志和业务 日志服务器，maxwell kafka
DWD 根据数据对象为单位进行分流，比如订单、页面访问等等。 flink kafka
DWM 对于部分数据对象进行进一步加工，比如独立访问、跳出行为。依旧是明细数据。 flink kafka
DIM 维度数据 flink hbase
DWS 根据某个维度主题将多个事实数据轻度聚合，形成主题宽表。 flink clickhouse
ADS 把 Clickhouse 中的数据根据可视化需要进行筛选聚合。 clickhouse，sql 可视化展示

目前进行到的阶段

3.Flink实时项目之流程分析及环境搭建

2. 环境搭建

https://github.com/zhangbaohpu/gmall-flink-parent

环境

  • jdk-1.8
  • flink-1.12
  • scala-2.12
  • hadoop-2.7.7

在项目中新建maven模块gmall-realtime，没有父工程，pom文件如下

点击查看代码
<?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">

    <modelVersion>4.0.0</modelVersion>

    <groupId>com.zhangbao.gmall</groupId>

    <artifactId>gmall-realtime</artifactId>

    <version>1.0-SNAPSHOT</version>

    <properties>

        <java.version>1.8</java.version>

        <maven.compiler.source>8</maven.compiler.source>

        <maven.compiler.target>8</maven.compiler.target>

        <flink.version>1.12.0</flink.version>

        <scala.version>2.12</scala.version>

        <hadoop.version>2.7.7</hadoop.version>

    </properties>

    <dependencies>

        <dependency>

            <groupId>org.apache.flink</groupId>

            <artifactId>flink-java</artifactId>

            <version>${flink.version}</version>

        </dependency>

        <dependency>

            <groupId>org.apache.flink</groupId>

            <artifactId>flink-streaming-java_${scala.version}</artifactId>

            <version>${flink.version}</version>

        </dependency>

        <dependency>

            <groupId>org.apache.flink</groupId>

            <artifactId>flink-connector-kafka_${scala.version}</artifactId>

            <version>${flink.version}</version>

        </dependency>

        <dependency>

            <groupId>org.apache.flink</groupId>

            <artifactId>flink-clients_${scala.version}</artifactId>

            <version>${flink.version}</version>

        </dependency>

        <dependency>

            <groupId>org.apache.flink</groupId>

            <artifactId>flink-cep_${scala.version}</artifactId>

            <version>${flink.version}</version>

        </dependency>

        <dependency>

            <groupId>org.apache.flink</groupId>

            <artifactId>flink-json</artifactId>

            <version>${flink.version}</version>

        </dependency>

        <dependency>

            <groupId>com.alibaba</groupId>

            <artifactId>fastjson</artifactId>

            <version>1.2.68</version>

        </dependency>

        <!--如果保存检查点到 hdfs 上，需要引入此依赖-->

        <dependency>

            <groupId>org.apache.hadoop</groupId>

            <artifactId>hadoop-client</artifactId>

            <version>${hadoop.version}</version>

        </dependency>

        <!--Flink 默认使用的是 slf4j 记录日志，相当于一个日志的接口,我们这里使用 log4j 作为具体的日志实现-->

        <dependency>

            <groupId>org.slf4j</groupId>

            <artifactId>slf4j-api</artifactId>

            <version>1.7.25</version>

        </dependency>

        <dependency>

            <groupId>org.slf4j</groupId>

            <artifactId>slf4j-log4j12</artifactId>

            <version>1.7.25</version>

        </dependency>

        <dependency>

            <groupId>org.apache.logging.log4j</groupId>

            <artifactId>log4j-to-slf4j</artifactId>

            <version>2.14.0</version>

        </dependency>

    </dependencies>

    <build>

        <plugins>

            <plugin>

                <groupId>org.apache.maven.plugins</groupId>

                <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>

        </plugins>

    </build>

</project>

项目结构

3.Flink实时项目之流程分析及环境搭建

log4j.properties

点击查看代码
log4j.rootLogger=warn,stdout

log4j.appender.stdout=org.apache.log4j.ConsoleAppender

log4j.appender.stdout.target=System.out

log4j.appender.stdout.layout=org.apache.log4j.PatternLayout

log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n

接下来的准备：

我们前面采集的日志数据已经保存到 Kafka 中，作为日志数据的 ODS 层，从 kafka 的ODS 层读取的日志数据分为 3 类, 页面日志、启动日志和曝光日志。这三类数据虽然都是用户行为数据，但是有着完全不一样的数据结构，所以要拆分处理。将拆分后的不同的日志写回 Kafka 不同主题中，作为日志 DWD 层。

相关文章