Java 和 Scala 实现 Spark RDD 转换成 DataFrame 的两种方法小结
在本文中，我们将讨论如何使用 Java 和 Scala 将 Spark RDD 转换成 DataFrame，並且介绍两种实现方法。
准备数据源
在项目下新建一个 student.txt 文件，内容如下：
1,zhangsan,20
2,lisi,21
3,wanger,19
4,fangliu,18
Java 版实现
第一步：新建一个 Student 的 Bean 对象，实现序列化和 toString() 方法。
```java
package com.cxd.sql;
import java.io.Serializable;
@SuppressWarnings("serial")
public class Student implements Serializable {
String sid;
String sname;
int sage;
public String getSid() {
return sid;
}
public void setSid(String sid) {
this.sid = sid;
}
public String getSname() {
return sname;
}
public void setSname(String sname) {
this.sname = sname;
}
public int getSage() {
return sage;
}
public void setSage(int sage) {
this.sage = sage;
}
@Override
public String toString() {
return "Student [sid=" + sid + ", sname=" + sname + ", sage=" + sage + "]";
}
}
```
第二步：将 RDD 转换成 DataFrame。
```java
package com.cxd.sql;
import java.util.ArrayList;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
public class TxtToParquetDemo {
public static void main(String[] args) {
SparkConf conf = new SparkConf().setAppName("TxtToParquet").setMaster("local");
SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
reflectTransform(spark);//Java 反射
dynamicTransform(spark);//动态转换
}
/
* 通过 Java 反射转换
* @param spark
*/
private static void reflectTransform(SparkSession spark) {
JavaRDD<String> source = spark.read().textFile("stuInfo.txt").javaRDD();
JavaRDD<Student> rowRDD = source.map(line -> {
String parts[] = line.split(",");
Student stu = new Student();
stu.setSid(parts[0]);
stu.setSname(parts[1]);
stu.setSage(Integer.parseInt(parts[2]));
return stu;
});
// 创建 StructType
StructType schema = DataTypes.createStructType(new StructField[] {
DataTypes.createStructField("sid", DataTypes.StringType, true),
DataTypes.createStructField("sname", DataTypes.StringType, true),
DataTypes.createStructField("sage", DataTypes.IntegerType, true)
});
// 将 JavaRDD 转换成 DataFrame
Dataset<Row> df = spark.createDataFrame(rowRDD, schema);
}
/
* 通过动态转换
* @param spark
*/
private static void dynamicTransform(SparkSession spark) {
// ...
}
}
```
Scala 版实现
在 Scala 中，我们可以使用 case class 来定义Student 对象，并使用 SparkSQL 的 implicits 将 RDD 转换成 DataFrame。
```scala
case class Student(sid: String, sname: String, sage: Int)
object TxtToParquetDemo {
def main(args: Array[String]) {
val spark = SparkSession.builder.appName("TxtToParquet").master("local").getOrCreate
val source = spark.read.textFile("stuInfo.txt").rdd
val rowRDD = source.map { line =>
val parts = line.split(",")
Student(parts(0), parts(1), parts(2).toInt)
}
import spark.implicits._
val df = rowRDD.toDF
}
}
```
结论
在本文中，我们讨论了如何使用 Java 和 Scala 将 Spark RDD 转换成 DataFrame，並且介绍了两种实现方法：Java 反射和动态转换。在实际应用中，我们可以根据不同的需求选择合适的实现方法。
- 1
- 2
前往页