Apache DataFu 快速入门与集成示例

本文面向首次接触 Apache DataFu 的工程师，用「环境 → 首例 → 解读」的路径快速跑通一个真实可运行的最小示例，并进一步给出在普通 Java 工程、Pig、Spark 三类场景中的集成方式。首例的全部命令与回显均来自真实执行。

一、环境准备

1.1 前置条件

JDK 8 或 11（实测 JDK 8、JDK 11 均可）。
Maven 3.5+（实测 3.5.4 / 3.6.3 均可用）。
能访问 Maven 中央仓库 https://repo1.maven.org/maven2/。

版本验证：

java -version
mvn -v

实测回显（Ubuntu 22.04 + JDK 11）：

openjdk version "11.0.32.1" 2026-08-18
OpenJDK 64-Bit Server VM (build 11.0.32.1+1-post-1ubuntu1-22.04-Ubuntu, mixed mode, sharing)
Apache Maven 3.6.3
Java version: 11.0.32.1, vendor: Ubuntu

1.2 关于 DataFu 坐标

核心库 + Pig UDF：org.apache.datafu:datafu-pig:1.6.1。
不存在 org.apache.datafu:datafu 核心构件（实测确认），请勿使用该坐标。
另有 datafu-hourglass:1.6.1、datafu-spark_2.12:2.1.0 等子模块。

二、第一个可运行示例

2.1 创建工程

mkdir -p /root/datafu-lab/src/main/java/demo
cd /root/datafu-lab

2.2 编写 pom.xml

<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>demo</groupId>
  <artifactId>datafu-lab</artifactId>
  <version>1.0</version>
  <properties>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>
  <dependencies>
    <dependency>
      <groupId>org.apache.datafu</groupId>
      <artifactId>datafu-pig</artifactId>
      <version>1.6.1</version>
    </dependency>
    <dependency>
      <groupId>org.apache.pig</groupId>
      <artifactId>pig</artifactId>
      <version>0.17.0</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-common</artifactId>
      <version>2.7.7</version>
    </dependency>
  </dependencies>
</project>

说明：datafu-pig 把 Pig/Hadoop 标为 provided，直接调用 UDF 需显式补 pig 与 hadoop-common（否则编译报找不到 org.apache.hadoop.io.WritableComparable）。

2.3 编写主类

文件 src/main/java/demo/DataFuDemo.java：

package demo;

import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;

import datafu.pig.hash.MD5;
import datafu.pig.hash.SHA;
import datafu.pig.util.Base64Encode;
import datafu.pig.sampling.ReservoirSample;

public class DataFuDemo {
  public static void main(String[] args) throws Exception {
    TupleFactory tf = TupleFactory.getInstance();
    BagFactory bf = BagFactory.getInstance();

    MD5 md5 = new MD5();
    System.out.println("MD5(apache-datafu)          = " + md5.exec(tf.newTuple("apache-datafu")));

    SHA sha = new SHA("256");
    System.out.println("SHA-256(apache-datafu)      = " + sha.exec(tf.newTuple("apache-datafu")));

    Base64Encode b64 = new Base64Encode();
    System.out.println("Base64Encode(datafu)        = " + b64.exec(tf.newTuple("datafu")));

    ReservoirSample rs = new ReservoirSample("2");
    DataBag input = bf.newDefaultBag();
    for (int i = 1; i <= 10; i++) {
      Tuple row = tf.newTuple(1);
      row.set(0, i);
      input.add(row);
    }
    Tuple group = tf.newTuple(1);
    group.set(0, input);
    rs.accumulate(group);
    DataBag sampled = rs.getValue();
    StringBuilder sb = new StringBuilder();
    for (Tuple t : sampled) {
      if (sb.length() > 0) sb.append(",");
      sb.append(t.get(0));
    }
    System.out.println("ReservoirSample(2 of 1..10) = [" + sb + "] size=" + sampled.size());
  }
}

2.4 编译与运行

mvn -q -B dependency:build-classpath -Dmdep.outputFile=cp.txt
mvn -B compile
java -Xmx384m -cp "target/classes:$(cat cp.txt)" demo.DataFuDemo

真实回显：

MD5(apache-datafu)          = f3faa986c556bc49f4131c987f22d04a
SHA-256(apache-datafu)      = 07ee215182b6749bc26d1afa300da5f2bf1833d8f0c5396e56a2f36110dafc56
Base64Encode(datafu)        = ZGF0YWZ1
ReservoirSample(2 of 1..10) = [1,9] size=2

三、首例解读

3.1 依赖与运行时

datafu-pig 提供全部 UDF 类；pig 提供 Tuple/DataBag/TupleFactory；hadoop-common 提供 WritableComparable 等基础类型。
用 TupleFactory/BagFactory 构造数据即可在纯 Java 中调用 UDF 的 exec / accumulate / getValue 方法，无需启动 Pig 引擎。

3.2 结果解读

MD5("apache-datafu") 与 SHA-256("apache-datafu") 为确定性摘要，两台机器输出一致，可用于去重键、指纹计算。
Base64Encode("datafu") 输出 ZGF0YWZ1，可用于字段编码传输。
ReservoirSample("2") 从 1..10 中抽 2 个元素，样本数恒为 2，元素随机（实测不同运行分别得到 [1,9]、[6,10]），适合单遍数据流的等量抽样。

3.3 常见易错点

Q1： new SHA("SHA-256") 报 SHA-SHA-256 MessageDigest not available。

A：该构造只接受算法后缀，写 new SHA("256")。

Q2： 采样输出出现全 null 元组（如 (,,,,,)）。

A： TupleFactory.newTuple(int) 是「创建 n 个字段的空元组」。应写 newTuple(1) 后 set(0, 值)，而不是 newTuple(值)。

四、集成示例

4.1 普通 Java 工程（作为工具库，实测验证）

在 ETL/批处理代码中把 DataFu 当作普通工具库使用，最适合哈希、采样、集合运算等纯计算场景：

import datafu.pig.hash.SHA;
import org.apache.pig.data.TupleFactory;

SHA sha = new SHA("256");
String fingerprint = sha.exec(TupleFactory.getInstance().newTuple("user-1024"));
// fingerprint = 07ee... 形式，可作为唯一键或去重依据

Maven 依赖：

<dependency>
  <groupId>org.apache.datafu</groupId>
  <artifactId>datafu-pig</artifactId>
  <version>1.6.1</version>
</dependency>
<dependency>
  <groupId>org.apache.pig</groupId>
  <artifactId>pig</artifactId>
  <version>0.17.0</version>
</dependency>
<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-common</artifactId>
  <version>2.7.7</version>
</dependency>

4.2 Hadoop / Pig 集成

在已安装 Pig 的集群中，把 DataFu jar 注册后即可在脚本里调用 UDF：

REGISTER /path/to/datafu-pig-1.6.1.jar;

DEFINE md5 datafu.pig.hash.MD5();
DEFINE rs  datafu.pig.sampling.ReservoirSample('2');
DEFINE med datafu.pig.stats.Median();

data  = LOAD '/data/events' AS (uid:chararray, amount:double);
hashed = FOREACH data GENERATE md5(uid) AS uid_hash, amount;
grouped = GROUP hashed ALL;
GENERATE_FOR_STATS = FOREACH grouped GENERATE med(hashed.amount);
DUMP GENERATE_FOR_STATS;

依赖通过 REGISTER 引入，无需修改 Pig 安装。
集群运行内存由 mapreduce.map.memory.mb / mapreduce.reduce.memory.mb 及对应 java.opts 控制，建议任务 JVM -Xmx 约为任务内存的 80%。

说明：本示例为标准集群用法；本文实测环境未安装 pig CLI，故未做引擎级运行验证，UDF 功能已由 4.1 的 Java 直接调用验证。

4.3 Spark 集成

Spark 工程使用 datafu-spark（按集群 Scala 版本选择 _2.12 或 _2.11）：

<dependency>
  <groupId>org.apache.datafu</groupId>
  <artifactId>datafu-spark_2.12</artifactId>
  <version>2.1.0</version>
  <scope>provided</scope>
</dependency>

Scala 侧调用示意：

import datafu.spark.DataFrameOps._

val spark = org.apache.spark.sql.SparkSession.builder
  .appName("DataFu-Spark-Demo")
  .master("local[*]")
  .getOrCreate()

import spark.implicits._
val df = Seq(("a", 1.0), ("b", 2.0), ("c", 3.0)).toDF("k", "v")
// 例如使用 DataFu 提供的聚合/降采样算子
df.show()

说明：datafu-spark_2.12:2.1.0 坐标与 jar 可用性已实测确认（HTTP 200）；上述 Scala 代码为集成示意，未在本文环境编译运行。

五、快速上手小结

5.1 最短路径

引入 org.apache.datafu:datafu-pig:1.6.1。
独立 Java 工程补 pig + hadoop-common。
用 TupleFactory 装配输入，调用 UDF 的 exec（哈希/编码）或 accumulate+getValue（采样/聚合）。
运行：java -Xmx384m -cp "target/classes:$(cat cp.txt)" <MainClass>。

5.2 能力索引

需求	推荐类
指纹/去重键	`datafu.pig.hash.MD5` / `SHA` / `Hasher`
等量随机抽样	`datafu.pig.sampling.ReservoirSample` / `SimpleRandomSample`
按权重抽样	`datafu.pig.sampling.WeightedReservoirSample`
中位数/分位数	`datafu.pig.stats.Median` / `Quantile` / `StreamingQuantile`
去重计数	`datafu.pig.stats.HyperLogLogPlusPlus`
集合交并差	`datafu.pig.sets.SetIntersect` / `SetUnion` / `SetDifference`
编码	`datafu.pig.util.Base64Encode` / `Base64Decode`