如何将List转换为JavaRDD

Question

我们知道，在Spark中有一个方法rdd.collect，它可以将RDD转换为列表。

我们知道，在spark中有一个名为rdd.collect的方法，可以将RDD转化为列表。

List<String> f= rdd.collect();
String[] array = f.toArray(new String[f.size()]);

我希望在项目中做完全相反的事情。我有一个String类型的ArrayList，想将其转换为JavaRDD。我已经寻找这个解决方案相当长的时间了，但还没有找到答案。请问有人可以帮帮我吗？

- Amitabh Ranjan

Answer 1

你正在寻找 JavaSparkContext.parallelize(List) 等类似功能。这与 Scala API 中的操作方式相同。

- Sean Owen

---

感谢Sean的帮助。它将列表转换为JavaRDD。 - Amitabh Ranjan

你了解Spark中的“JdbcRDD”吗？我正在尝试在Java中实现它，但是没有找到任何关于它的文章或示例。我有一个链接（https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala），但是我无法理解其中的Scala代码。 - Amitabh Ranjan

@AmitabhRanjan 需要另外一个问题来解决这个问题。 - samthebest

Answer 2

补充Sean Owen和其他人的解决方案

您可以使用JavaSparkContext#parallelizePairs处理Tuple的List

List<Tuple2<Integer, Integer>> pairs = new ArrayList<>();
pairs.add(new Tuple2<>(0, 5));
pairs.add(new Tuple2<>(1, 3));

JavaSparkContext sc = new JavaSparkContext();

JavaPairRDD<Integer, Integer> rdd = sc.parallelizePairs(pairs);

- mrsrinivas

Answer 3

有两种方法可以将集合转换为RDD。

1) sc.Parallelize(collection)
2) sc.makeRDD(collection)

这两种方法是相同的，因此我们可以使用任何一种

- Abhash Kumar

---

+1 -- 这是 Databricks 的完美解决方案。虽然我认为 sc 关键字在该上下文之外没有意义？ - Paul

我们在spark-shell中创建了一个SparkContext对象（sc），否则我们必须创建一个SparkContext（每个程序只能有一个SparkContext）。 - Abhash Kumar

哦，好的，所以它比我想象的还要更通用。我戴着Databricks眼镜。 - Paul

Answer 4

如果您正在使用.scala文件，或者您不想或无法使用JavaSparkContext，那么您可以：

1. 使用SparkContext代替JavaSparkContext
2. 将您的Java List转换为Scala List
3. 使用SparkContext的parallelize方法

例如：

List<String> javaList = new ArrayList<>()
javaList.add("abc")
javaList.add("def")
sc.parallelize(javaList.asScala)

这将为您生成一个RDD。

- malvadao

Answer 5

List<StructField> fields = new ArrayList<>();
fields.add(DataTypes.createStructField("fieldx1", DataTypes.StringType, true));
fields.add(DataTypes.createStructField("fieldx2", DataTypes.StringType, true));
fields.add(DataTypes.createStructField("fieldx3", DataTypes.LongType, true));


List<Row> data = new ArrayList<>();
data.add(RowFactory.create("","",""));
Dataset<Row> rawDataSet = spark.createDataFrame(data, schema).toDF();

- Mantas