请求协助开发Pig Java UDF：从元组生成目标Bag结构

嘿，作为Java新手能想到用UDF处理这类数据真不错！我帮你整理了完整的实现方案，包括普通Java文件处理程序和**Hive UDTF（表生成函数）**两种场景，每个步骤都写得很详细，适合刚入门的你：

先明确需求拆解 #

我们要做的核心事情很清晰：

• 读取多个文本文件里的每行数据（格式比如Montreal, 5 3 10 9 8）
• 把每行拆成「城市名」和「一串数字」两部分
• 给每个数字都配上对应的城市，变成(城市, 数字)的结构
• 最后按城市分组，输出你要求的Bag格式

场景1：普通Java程序处理本地文件 #

如果你只是想读取本地文本文件并输出结果，下面是可直接运行的代码，注释超详细：

import java.io.File;
import java.io.FileNotFoundException;
import java.util.*;

public class CityDataProcessor {
    public static void main(String[] args) {
        // 这里替换成你的三个文本文件的实际路径，可以是相对路径或绝对路径
        List<String> filePaths = Arrays.asList(
                "montreal_data.txt",
                "toronto_data.txt",
                "edmonton_data.txt"
        );

        // 用Map存每个城市对应的所有(城市,数字)对，键是城市名，值是配对后的字符串列表
        Map<String, List<String>> cityDataMap = new HashMap<>();

        // 逐个读取文件
        for (String filePath : filePaths) {
            // try-with-resources语法：自动关闭文件流，新手不用手动关，避免资源泄漏
            try (Scanner scanner = new Scanner(new File(filePath))) {
                // 逐行读内容
                while (scanner.hasNextLine()) {
                    String line = scanner.nextLine().trim();
                    // 跳过空行，避免报错
                    if (line.isEmpty()) continue;

                    // 按「逗号+任意空格」拆分城市和数字部分，适配你给的格式
                    String[] parts = line.split(",\\s+");
                    if (parts.length != 2) {
                        System.err.println("这行格式不对，跳过：" + line);
                        continue;
                    }

                    String city = parts[0];
                    String numbersStr = parts[1];
                    // 把数字串按空格拆成单个数字
                    String[] numberArray = numbersStr.split("\\s+");

                    // 给每个数字配对城市，加入Map
                    for (String num : numberArray) {
                        // 简单校验数字有效性，避免非数字内容报错
                        if (!num.matches("\\d+")) {
                            System.err.println("数字格式不对，跳过：" + num);
                            continue;
                        }
                        String pair = String.format("(%s,%s)", city, num);
                        // 如果Map里还没这个城市，先创建空列表；然后把配对结果加进去
                        cityDataMap.computeIfAbsent(city, k -> new ArrayList<>()).add(pair);
                    }
                }
            } catch (FileNotFoundException e) {
                System.err.println("找不到这个文件哦：" + filePath);
                e.printStackTrace();
            }
        }

        // 按要求的格式输出结果
        for (Map.Entry<String, List<String>> entry : cityDataMap.entrySet()) {
            String city = entry.getKey();
            List<String> pairs = entry.getValue();
            // 把列表拼成你要的Bag格式：{(城市,数字),(城市,数字)...}
            String bagStr = "{" + String.join(",", pairs) + "}";
            System.out.println(bagStr);
        }
    }
}

运行说明： #

1. 把你的三个文本文件放到项目根目录，或者修改filePaths里的路径为绝对路径
2. 编译运行这个类，就能得到你想要的输出啦

场景2：Hive UDTF（表生成函数） #

如果你是要在Hive里用UDF处理这类数据，那需要写UDTF（表生成函数），因为要把一行数据拆成多行输出。下面是实现代码：

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

import java.util.ArrayList;
import java.util.List;

public class CityNumberUDTF extends GenericUDTF {
    @Override
    public StructObjectInspector initialize(ObjectInspector[] args) throws UDFArgumentException {
        // 检查输入参数：必须传两个参数，第一个是城市名，第二个是空格分隔的数字串
        if (args.length != 2) {
            throw new UDFArgumentException("需要传两个参数：城市名、空格分隔的数字字符串");
        }

        // 定义输出的结构：两个字段，city（字符串）和number（整数）
        List<String> fieldNames = new ArrayList<>();
        List<ObjectInspector> fieldOIs = new ArrayList<>();
        fieldNames.add("city");
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        fieldNames.add("number");
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaIntObjectInspector);

        return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);
    }

    @Override
    public void process(Object[] args) throws HiveException {
        String city = args[0].toString();
        String numbersStr = args[1].toString();
        String[] numberArray = numbersStr.split("\\s+");

        // 遍历每个数字，输出一行(城市,数字)
        for (String numStr : numberArray) {
            try {
                int number = Integer.parseInt(numStr);
                Object[] output = new Object[2];
                output[0] = city;
                output[1] = number;
                // 把结果输出到Hive表中
                forward(output);
            } catch (NumberFormatException e) {
                // 遇到非数字内容就跳过，不报错
                continue;
            }
        }
    }

    @Override
    public void close() throws HiveException {
        // 这里不需要额外操作，空实现即可
    }
}

Hive中使用步骤： #

1. 把代码打包成JAR文件（比如CityNumberUDTF.jar）
2. 打开Hive客户端，添加JAR：

   ADD JAR /path/to/your/jar/CityNumberUDTF.jar;
   

3. 创建临时函数：

   CREATE TEMPORARY FUNCTION split_city_number AS 'com.yourpackage.CityNumberUDTF';
   

   注意把com.yourpackage改成你实际的包名
4. 使用函数（假设你的表叫city_data，有city和numbers两个字段）：

   -- 直接拆分成每行一个(城市,数字)
   SELECT split_city_number(city, numbers) FROM city_data;
   
   -- 如果要按城市分组生成你要的Bag格式，用collect_list
   SELECT 
       city, 
       collect_list(named_struct('city', city, 'number', number)) AS city_bag
   FROM (
       SELECT split_city_number(city, numbers) AS (city, number) 
       FROM city_data
   ) t
   GROUP BY city;
   

内容的提问来源于stack exchange，提问作者HasanSalah