flink1.12.x Transformation-map算子

作者：马育民 • • 阅读：10082

勿动，被 flink1.12.x Transformation-filter算子 引用

说明

将 数据流 中的 一个数据 进行 转换，生成一个新数据

注意：

  • 经过 map算子 转换后，数据的 数量不变
  • 可设置并行度

api

map() 方法

通过 source 调用 map() 方法，map() 方法如下：

SingleOutputStreamOperator<R> map(MapFunction<T, R> mapper)

需要传入 MapFunction 类对象

MapFunction

源码如下：

@FunctionalInterface // 说明支持lambda表达式
@Public
public interface MapFunction<T, O> extends Function, Serializable {
    O map(T var1) throws Exception;
}

案例

从kafka接收数据，数据格式是 json格式，不方便做处理，所以将其转成 java对象（UserLog），便于后续操作处理

pom.xml

<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.12.0</flink.version>
</properties>

<dependencies>
    <!--java-->
    <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_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>1.7.30</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.30</version>
    </dependency>
    <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>1.2.17</version>
    </dependency>
    <!-- kafka Dependency -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
</dependencies>

UserLog 封装类

package filter;

import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonFormat;

import java.util.Date;

/**
 * 用户日志封装类
 */
public class UserLog {
    private String name;
    private String url;
    // 下面注解，会自动将字符串格式的日期转成 Date对象
    // 需要指定时区，否则差8小时
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss",timezone = "GMT+8")
    private Date time;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getUrl() {
        return url;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public Date getTime() {
        return time;
    }

    public void setTime(Date time) {
        this.time = time;
    }

    @Override
    public String toString() {
        return "UserLog{" +
                "name='" + name + '\'' +
                ", url='" + url + '\'' +
                ", time=" + time +
                '}';
    }
}

主类

传统方式实现Map类，即：实现 MapFunction 接口

package filter;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

public class TestMap {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 定义kafka相关配置项
        Properties pro = new Properties();
        pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop1:9092,hadoop2:9092,hadoop3:9092");
        pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"consumer-group");
        pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
        // 内置连接器，创建FlinkKafka消费者，需要指定主题、反序列化、kafka配置
        FlinkKafkaConsumer<String> fkc = new FlinkKafkaConsumer<>("kafka2flink", new SimpleStringSchema(), pro);
        DataStreamSource<String> dss = env.addSource(fkc);

        SingleOutputStreamOperator<UserLog> soso = dss.map(new MyMap());
        soso.print();

        env.execute();
    }

    /**
     * 第一个泛型，表示原数据的类型，从kafka接收到的数据，即：json格式的字符串
     * 第二个泛型，表示转换后的数据类型，即：UserLogin
     */
    public static class MyMap implements MapFunction<String,UserLog>{
        // 形参就是从kafka接收到的数据，即：json格式的字符串
        @Override
        public UserLog map(String s) throws Exception {
            // 通过jackson将 json字符串转换UserLog对象
            ObjectMapper mapper = new ObjectMapper();
            UserLog result = mapper.readValue(s, UserLog.class);
            return result;
        }
    }
}

启动zk

登录 hadoop1，执行下面命令：

/program/bin/zk.sh start

启动kafka

登录 hadoop1，执行下面命令：

/program/bin/kafka.sh start

启动 flink

在 idea 中运行 flink 程序

创建生产者、指定主题

登录 hadoop1，执行下面命令：

/program/kafka_2.12-2.4.1/bin/kafka-console-producer.sh --broker-list hadoop1:9092,hadoop2:9092,hadoop3:9092 --topic kafka2flink

然后输入：

{"name":"李雷","url":"/jianpan.html","time":"2020-09-01 10:00:00"}
{"name":"lucy","url":"/lv.html","time":"2020-09-01 12:00:00"}
{"name":"lucy","url":"/dior.html","time":"2020-09-01 15:00:00"}

执行结果

在 idea 控制台中打印结果如下：

5> UserLog{name='����', url='/jianpan.html', time=Tue Sep 01 18:00:00 CST 2020}
5> UserLog{name='lucy', url='/lv.html', time=Tue Sep 01 20:00:00 CST 2020}
5> UserLog{name='lucy', url='/dior.html', time=Tue Sep 01 23:00:00 CST 2020}

主类-lambda表达式实现

由于 MapFunction 接口有 @FunctionalInterface 注解，可知是支持 lambda 表达式的

实现方式如下：

public class TestMap {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 定义kafka相关配置项
        Properties pro = new Properties();
        pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop1:9092,hadoop2:9092,hadoop3:9092");
        pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"consumer-group");
        pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
        // 内置连接器，创建FlinkKafka消费者，需要指定主题、反序列化、kafka配置
        FlinkKafkaConsumer<String> fkc = new FlinkKafkaConsumer<>("kafka2flink", new SimpleStringSchema(), pro);
        DataStreamSource<String> dss = env.addSource(fkc);

        /* lambda表达式实现
         * 第一个泛型，表示原数据的类型
         * 第二个泛型，表示转换后的数据类型
         */
        SingleOutputStreamOperator<UserLog> soso = dss.map( s -> {
                ObjectMapper mapper = new ObjectMapper();
                UserLog result = mapper.readValue(s, UserLog.class);
                return result;
            }
        );
        soso.print();

        env.execute();
    }


}

原文出处：/show_1IX435Jamyv3.html

吉ICP备15003043号