使用 Spark 将字段添加到 Csv

Question

所以，我有一个包含空间 ( latitude, longitude) 和时间 ( timestamp) 数据的 CSV 。

为了对我们有用，我们将空间信息转换为“ geohash”，将时间信息转换为“ timehash”。

问题是，如何使用 spark 为 CSV 中的每一行添加geohash和timehash作为字段（因为数据大约为200 GB）？

我们尝试使用JavaPairRDD它的功能mapTopair，但问题仍然在于如何转换回 aJavaRdd然后转换为 CSV？所以我认为这是一个糟糕的解决方案，我要求一个简单的方法。

问题更新：

在@Alvaro 得到帮助后，我创建了这个 java 类：

public class Hash {

public static SparkConf Spark_Config;

public static JavaSparkContext Spark_Context;

UDF2 geohashConverter = new UDF2<Long, Long, String>() {

    

    public String call(Long latitude, Long longitude) throws Exception {

        // convert here

        return "calculate_hash";

    }

};

UDF1 timehashConverter = new UDF1<Long, String>() {

    

    public String call(Long timestamp) throws Exception {

        // convert here

        return "calculate_hash";

    }

};

public Hash(String path) {

    SparkSession spark = SparkSession

            .builder()

            .appName("Java Spark SQL Example")

            .config("spark.master", "local")

            .getOrCreate();

    

    spark.udf().register("geohashConverter", geohashConverter, DataTypes.StringType);

    spark.udf().register("timehashConverter", timehashConverter, DataTypes.StringType);

    

Dataset df=spark.read().csv(path)

    .withColumn("geohash", callUDF("geohashConverter", col("_c6"), col("_c7")))

    .withColumn("timehash", callUDF("timehashConverter", col("_c1")))

.write().csv("C:/Users/Ahmed/Desktop/preprocess2");

 }

public static void main(String[] args) {

    String path = "C:/Users/Ahmed/Desktop/cabs_trajectories/cabs_trajectories/green/2013";

    Hash h = new Hash(path);

}

}

然后我得到序列化问题，当我删除时消失 write().csv()