基于Flink JAR的数据处理

例如Flink Java代码“WindowJoin.java”，完整示例如下：

package org.apache.flink.streaming.examples.join;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import java.time.Duration;
public class WindowJoin {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        String outputPath = null;
        for (int i = 0; i < args.length - 1; i++) {
            if ("--output".equals(args[i])) {
                outputPath = args[i + 1];
                break;
            }
        }
        DataStream<Tuple3<String, Integer, Long>> orderStream = env
            .fromElements(
                Tuple3.of("order1", 100, 1000L),
                Tuple3.of("order2", 200, 2000L),
                Tuple3.of("order3", 150, 3000L),
                Tuple3.of("order4", 300, 4000L)
            )
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<Tuple3<String, Integer, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                    .withTimestampAssigner((event, ts) -> event.f2)
            );
        DataStream<Tuple3<String, String, Long>> paymentStream = env
            .fromElements(
                Tuple3.of("order1", "credit", 1500L),
                Tuple3.of("order2", "debit", 2500L),
                Tuple3.of("order3", "credit", 3500L),
                Tuple3.of("order4", "debit", 4500L)
            )
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<Tuple3<String, String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                    .withTimestampAssigner((event, ts) -> event.f2)
            );
        DataStream<String> result = orderStream
            .join(paymentStream)
            .where(order -> order.f0)
            .equalTo(payment -> payment.f0)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .apply(
                (JoinFunction<Tuple3<String, Integer, Long>, Tuple3<String, String, Long>, String>)
                    (order, payment) ->
                        "orderId=" + order.f0
                            + ", amount=" + order.f1
                            + ", payMethod=" + payment.f1
            );
        if (outputPath != null) {
            FileSink<String> sink = FileSink
                .forRowFormat(new Path(outputPath), new SimpleStringEncoder<String>("UTF-8"))
                .withRollingPolicy(OnCheckpointRollingPolicy.build())
                .build();
            result.sinkTo(sink);
        } else {
            result.print();
        }
        env.execute("WindowJoin QuickStart");
    }
}

Flink Java的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">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>window-join</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>jar</packaging>

    <properties>
        <maven.compiler.source>11</maven.compiler.source>
        <maven.compiler.target>11</maven.compiler.target>
        <flink.version>1.20.0</flink.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-files</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.5.0</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>org.apache.flink.streaming.examples.join.WindowJoin</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>