Apache Wayang 快速入门与集成示例

一、Wayang 与核心概念

Apache Wayang 是跨平台数据处理引擎：只写一遍处理逻辑，通过注册不同后端插件，把同一计划跑在 Java Stream、Spark、Flink 或数据库上，也可以交给代价优化器自动选择执行位置。本文在 OpenJDK 17.0.20.1 + Maven 3.6.3 环境对 1.1.1 版本完成全部示例实测。

四个核心概念：

概念	类	作用
上下文	org.apache.wayang.core.api.WayangContext	持有 Configuration 并注册后端插件
插件	org.apache.wayang.java.Java / org.apache.wayang.spark.Spark	声明可用的执行后端，如 Java.basicPlugin()
构建器	org.apache.wayang.api.JavaPlanBuilder	fluent 风格装配读、转换、写算子
元组	org.apache.wayang.basic.data.Tuple2	Key-Value 型中间数据

前置条件：按《Ubuntu 22.04 安装 Apache Wayang 指南》或《CentOS 8 安装 Apache Wayang 指南》完成 JDK 17、Maven 安装，并准备输入文件：

mkdir -p /root/wayang-demo/src/main/java/com/example /root/wayang-data
cat > /root/wayang-data/input.txt << 'EOF'
Apache Wayang is a cross platform data processor
wayang can run on java streams and spark
Apache Wayang mixes java and spark backends
big data with wayang big data
EOF

二、完整工程搭建

2.1 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>com.example</groupId>
    <artifactId>wayang-demo</artifactId>
    <version>1.0.0</version>
    <packaging>jar</packaging>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
        <wayang.version>1.1.1</wayang.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.wayang</groupId>
            <artifactId>wayang-core</artifactId>
            <version>${wayang.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.wayang</groupId>
            <artifactId>wayang-basic</artifactId>
            <version>${wayang.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.wayang</groupId>
            <artifactId>wayang-java</artifactId>
            <version>${wayang.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.wayang</groupId>
            <artifactId>wayang-api-scala-java</artifactId>
            <version>${wayang.version}</version>
        </dependency>
    </dependencies>
</project>

wayang-api-scala-java 是 JavaPlanBuilder 所在模块，1.x 必须显式引入，缺失会报 package org.apache.wayang.api does not exist（实测踩坑）。

2.2 工程目录结构

/root/wayang-demo
├── pom.xml
└── src/main/java/com/example
    ├── WordCount.java
    ├── WordCountFile.java
    ├── FilterAggregate.java
    └── LowLevelWordCount.java

2.3 WordCount 主示例

package com.example;

import org.apache.wayang.api.JavaPlanBuilder;
import org.apache.wayang.basic.data.Tuple2;
import org.apache.wayang.core.api.Configuration;
import org.apache.wayang.core.api.WayangContext;
import org.apache.wayang.java.Java;

import java.util.Arrays;
import java.util.Collection;

public class WordCount {
    public static void main(String[] args) {
        WayangContext wayang = new WayangContext(new Configuration())
                .withPlugin(Java.basicPlugin());

        Collection<Tuple2<String, Integer>> result = new JavaPlanBuilder(wayang)
                .withJobName("WordCount")
                .withUdfJarOf(WordCount.class)
                .readTextFile("file:///root/wayang-data/input.txt")
                .flatMap(line -> Arrays.asList(line.toLowerCase().split("\\W+")))
                .filter(word -> !word.isEmpty())
                .map(word -> new Tuple2<>(word, 1))
                .reduceByKey(Tuple2::getField0,
                        (t1, t2) -> new Tuple2<>(t1.getField0(), t1.getField1() + t2.getField1()))
                .collect();

        result.forEach(t -> System.out.println(t.getField0() + ": " + t.getField1()));
    }
}

编译并运行（堆限定 256-512 MB）：

cd /root/wayang-demo
export MAVEN_OPTS="-Xms256m -Xmx512m"
mvn package dependency:copy-dependencies -DoutputDirectory=target/lib
java -Xms256m -Xmx512m -cp "target/wayang-demo-1.0.0.jar:target/lib/*" com.example.WordCount

实测回显：

big: 2
a: 1
data: 3
streams: 1
cross: 1
run: 1
is: 1
wayang: 4
backends: 1
processor: 1
platform: 1
with: 1
can: 1
java: 2
apache: 2
spark: 2
and: 2
mixes: 1
on: 1

三、文件读写示例

把 collect 换成 writeTextFile 写回文件（注意 1.1.1 为三参数，第三位是字符编码，官方旧示例的两参数写法编译不过）：

package com.example;

import org.apache.wayang.api.JavaPlanBuilder;
import org.apache.wayang.basic.data.Tuple2;
import org.apache.wayang.core.api.Configuration;
import org.apache.wayang.core.api.WayangContext;
import org.apache.wayang.java.Java;

import java.util.Arrays;

public class WordCountFile {
    public static void main(String[] args) {
        WayangContext wayang = new WayangContext(new Configuration())
                .withPlugin(Java.basicPlugin());

        new JavaPlanBuilder(wayang)
                .withJobName("WordCountFile")
                .withUdfJarOf(WordCountFile.class)
                .readTextFile("file:///root/wayang-data/input.txt")
                .flatMap(line -> Arrays.asList(line.toLowerCase().split("\\W+")))
                .filter(word -> !word.isEmpty())
                .map(word -> new Tuple2<>(word, 1))
                .reduceByKey(Tuple2::getField0,
                        (t1, t2) -> new Tuple2<>(t1.getField0(), t1.getField1() + t2.getField1()))
                .writeTextFile("file:///root/wayang-data/output.txt",
                        t -> t.getField0() + ": " + t.getField1(), "UTF-8");
    }
}

运行并查看结果文件：

mvn -q package
java -Xms256m -Xmx512m -cp "target/wayang-demo-1.0.0.jar:target/lib/*" com.example.WordCountFile
sort /root/wayang-data/output.txt | head -5

实测回显：

a: 1
and: 2
apache: 2
backends: 1
big: 2

四、过滤聚合示例

集合数据源 + filter/map/reduce 的典型组合：

package com.example;

import org.apache.wayang.api.JavaPlanBuilder;
import org.apache.wayang.core.api.Configuration;
import org.apache.wayang.core.api.WayangContext;
import org.apache.wayang.java.Java;

import java.util.Arrays;
import java.util.Collection;

public class FilterAggregate {
    public static void main(String[] args) {
        WayangContext wayang = new WayangContext(new Configuration())
                .withPlugin(Java.basicPlugin());

        Collection<Integer> result = new JavaPlanBuilder(wayang)
                .withJobName("FilterAggregate")
                .loadCollection(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
                .filter(n -> n % 2 == 0)
                .map(n -> n * 10)
                .reduce((a, b) -> a + b)
                .collect();

        System.out.println("sum = " + result);
    }
}

实测运行：

java -Xms256m -Xmx512m -cp "target/wayang-demo-1.0.0.jar:target/lib/*" com.example.FilterAggregate

回显：

sum = [300]

五、底层算子 API 示例

不经过 JavaPlanBuilder，直接用 wayang-basic 算子拼装计划并执行（适合需要精细控制算子的场景）：

package com.example;

import org.apache.wayang.basic.data.Tuple2;
import org.apache.wayang.basic.operators.FilterOperator;
import org.apache.wayang.basic.operators.FlatMapOperator;
import org.apache.wayang.basic.operators.LocalCallbackSink;
import org.apache.wayang.basic.operators.MapOperator;
import org.apache.wayang.basic.operators.ReduceByOperator;
import org.apache.wayang.basic.operators.TextFileSource;
import org.apache.wayang.core.api.WayangContext;
import org.apache.wayang.core.plan.wayangplan.WayangPlan;
import org.apache.wayang.java.Java;

import java.util.Arrays;

public class LowLevelWordCount {
    @SuppressWarnings("unchecked")
    public static void main(String[] args) {
        Class<Tuple2<String, Integer>> tupleClass = (Class<Tuple2<String, Integer>>) (Class<?>) Tuple2.class;

        WayangContext wayang = new WayangContext().withPlugin(Java.basicPlugin());

        TextFileSource source = new TextFileSource("file:///root/wayang-data/input.txt");
        FlatMapOperator<String, String> tokens = new FlatMapOperator<>(
                line -> Arrays.asList(line.toLowerCase().split("\\W+")),
                String.class, String.class);
        FilterOperator<String> nonEmpty = new FilterOperator<>(
                w -> !w.isEmpty(), String.class);
        MapOperator<String, Tuple2<String, Integer>> pairs = new MapOperator<>(
                word -> new Tuple2<>(word, 1), String.class, tupleClass);
        ReduceByOperator<Tuple2<String, Integer>, String> counts = new ReduceByOperator<>(
                Tuple2::getField0,
                (t1, t2) -> new Tuple2<>(t1.getField0(), t1.getField1() + t2.getField1()),
                String.class, tupleClass);
        LocalCallbackSink<Tuple2<String, Integer>> sink = LocalCallbackSink.createStdoutSink(tupleClass);

        source.connectTo(0, tokens, 0);
        tokens.connectTo(0, nonEmpty, 0);
        nonEmpty.connectTo(0, pairs, 0);
        pairs.connectTo(0, counts, 0);
        counts.connectTo(0, sink, 0);

        wayang.execute(new WayangPlan(sink));
    }
}

底层 API 的两个签名要点（实测编译确认）：connectTo(输出端口, 目标算子, 输入端口)，端口在前；FlatMap/Filter 构造器 lambda 在前、Class 在后。运行：

java -Xms256m -Xmx512m -cp "target/wayang-demo-1.0.0.jar:target/lib/*" com.example.LowLevelWordCount

实测回显（节选，Tuple2 的 toString 格式）：

(big, 2)
(a, 1)
(data, 3)
(streams, 1)
(cross, 1)

六、双后端对比：Java 与 Spark local

6.1 集成 wayang-spark

在另一个工程（或同一工程追加依赖）引入 wayang-spark，并排除其传递的 hadoop-client：

<dependency>
    <groupId>org.apache.wayang</groupId>
    <artifactId>wayang-spark</artifactId>
    <version>1.1.1</version>
    <exclusions>
        <exclusion>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
        </exclusion>
    </exclusions>
</dependency>

为什么必须排除（实测定位过程）：wayang-spark 1.1.1 传递引入未 shaded 的 hadoop 3.3.6 全家桶，而 Spark 3.4.4 自带 shaded 的 hadoop-client-api/runtime 3.3.4，两套同名类混在 classpath 上时运行报错：

java.lang.NoSuchMethodError: 'org.apache.hadoop.fs.FSBuilder org.apache.hadoop.fs.FutureDataInputStreamBuilder.optLong(java.lang.String, long)'

排除后 classpath 只剩 Spark 自带的一套（hadoop-client-api-3.3.4.jar、hadoop-client-runtime-3.3.4.jar），冲突消失。

6.2 同一计划切换到 Spark

与 2.3 节相比只有三处差异——导入、插件注册、spark.master 注入，处理逻辑零改动：

package com.example;

import org.apache.wayang.api.JavaPlanBuilder;
import org.apache.wayang.basic.data.Tuple2;
import org.apache.wayang.core.api.Configuration;
import org.apache.wayang.core.api.WayangContext;
import org.apache.wayang.spark.Spark;

import java.util.Arrays;
import java.util.Collection;

public class WordCountSpark {
    public static void main(String[] args) {
        Configuration configuration = new Configuration();
        configuration.setProperty("spark.master", "local[2]");
        WayangContext wayang = new WayangContext(configuration)
                .withPlugin(Spark.basicPlugin());

        Collection<Tuple2<String, Integer>> result = new JavaPlanBuilder(wayang)
                .withJobName("WordCountSpark")
                .withUdfJarOf(WordCountSpark.class)
                .readTextFile("file:///root/wayang-data/input.txt")
                .flatMap(line -> Arrays.asList(line.toLowerCase().split("\\W+")))
                .filter(word -> !word.isEmpty())
                .map(word -> new Tuple2<>(word, 1))
                .reduceByKey(Tuple2::getField0,
                        (t1, t2) -> new Tuple2<>(t1.getField0(), t1.getField1() + t2.getField1()))
                .collect();

        result.forEach(t -> System.out.println("SPARK>> " + t.getField0() + ": " + t.getField1()));
    }
}

6.3 Spark local 运行

Java 17 上必须追加 --add-opens（否则报 IllegalAccessError ... sun.nio.ch.DirectBuffer），实测完整命令：

java -Xms256m -Xmx512m \
  --add-opens java.base/java.lang=ALL-UNNAMED \
  --add-opens java.base/java.lang.invoke=ALL-UNNAMED \
  --add-opens java.base/java.lang.reflect=ALL-UNNAMED \
  --add-opens java.base/java.io=ALL-UNNAMED \
  --add-opens java.base/java.net=ALL-UNNAMED \
  --add-opens java.base/java.nio=ALL-UNNAMED \
  --add-opens java.base/java.util=ALL-UNNAMED \
  --add-opens java.base/java.util.concurrent=ALL-UNNAMED \
  --add-opens java.base/java.util.concurrent.atomic=ALL-UNNAMED \
  --add-opens java.base/sun.nio.ch=ALL-UNNAMED \
  --add-opens java.base/sun.util.calendar=ALL-UNNAMED \
  --add-opens java.security.jgss/sun.security.krb5=ALL-UNNAMED \
  -cp "target/wayang-spark-demo-1.0.0.jar:target/lib/*" com.example.WordCountSpark

实测回显（节选，共 19 行）：

SPARK>> can: 1
SPARK>> big: 2
SPARK>> cross: 1
SPARK>> spark: 2
SPARK>> a: 1
SPARK>> is: 1

6.4 输出对比

对比项	Java 后端（WordCount）	Spark local 后端（WordCountSpark）
处理逻辑	完全相同	完全相同
代码差异	无	导入 Spark、注册 Spark.basicPlugin()、注入 spark.master 共三处
词频结果	wayang:4、data:3、big:2、spark:2 等	与 Java 后端完全一致（19 个词条相同）
外部依赖	仅 JDK	Spark 3.4.4 依赖约 1 GB（Maven 缓存）
启动开销	亚秒级	有 local 模式 JVM 初始化开销

结论：同一计划在两个后端输出一致，Wayang 的"一次编写、多端执行"在 local 单进程场景下可完整跑通。

七、JVM 堆验证

任务运行期间另开会话执行：

ps -ef | grep "[j]ava -Xms"

实测回显：

root        5820    5818 99 14:48 ?        00:00:02 java -Xms256m -Xmx512m -cp target/wayang-demo-1.0.0.jar:target/lib/* com.example.WordCount

jps -v

实测回显：

5820 WordCount -Xms256m -Xmx512m

两处回显确认 -Xms256m -Xmx512m 生效。Maven 构建期的堆由 MAVEN_OPTS 控制（推荐写入 /etc/profile.d/wayang-maven.sh），应用运行期的堆由命令行或 JAVA_OPTS 控制。

八、常见问题

**Q1：**编译报 package org.apache.wayang.api does not exist？补上 wayang-api-scala-java 依赖（版本与 wayang-core 同为 1.1.1）。1.x 把 JavaPlanBuilder 移到了该模块，实测补依赖后编译通过。

**Q2：**Spark 后端报 NoSuchMethodError，方法签名涉及 org.apache.hadoop.fs.FutureDataInputStreamBuilder？ hadoop 依赖冲突：排除 wayang-spark 传递的 org.apache.hadoop:hadoop-client，让 Spark 自带的 hadoop-client-api/runtime 3.3.4 生效，实测排除后运行正常。

**Q3：**Spark 后端报 IllegalAccessError ... cannot access class sun.nio.ch.DirectBuffer？ Java 17 模块封装限制，为 JVM 追加 6.3 节列出的 --add-opens 参数后实测跑通。

Q4：writeTextFile 编译报 no suitable method found？ 1.1.1 的签名是三参数 (uri, formatter, encoding)，补上编码实参如 "UTF-8" 即可，实测通过。

**Q5：**引入 Spark 依赖后构建很慢？ wayang-spark 连带 Spark 3.4.4 与 hadoop 依赖，Maven 首次下载约 1 GB，属正常现象；java 后端工程（约 150 MB）不受影响，可分别建工程管理。

九、参考链接

Apache Wayang 官网：https://wayang.apache.org/
快速上手指南：https://wayang.apache.org/docs/guide/getting-started
GitHub 仓库（apache/wayang）：https://github.com/apache/wayang
Maven Central 上 wayang-core 构件：https://repo1.maven.org/maven2/org/apache/wayang/wayang-core/
Maven Central 上 wayang-spark 构件：https://repo1.maven.org/maven2/org/apache/wayang/wayang-spark/
Sonatype Central（版本检索）：https://central.sonatype.com/artifact/org.apache.wayang/wayang-core