如何在Spark Java中使用StructType模式从JavaRDD<String>中读取csv格式的数据

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

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

public class CsvToDataFrame {
    public static void main(String[] args) {
        // 初始化Spark配置和上下文
        SparkConf conf = new SparkConf().setAppName("CsvToDataFrame").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);

        // 假设我们有一个JavaRDD<String>，其中包含CSV格式的数据
        JavaRDD<String> csvData = sc.textFile("path/to/your/csvfile.csv");

        // 定义CSV数据的Schema
        List<StructField> fields = new ArrayList<>();
        fields.add(DataTypes.createStructField("column1", DataTypes.StringType, true));
        fields.add(DataTypes.createStructField("column2", DataTypes.IntegerType, true));
        fields.add(DataTypes.createStructField("column3", DataTypes.DoubleType, true));
        StructType schema = DataTypes.createStructType(fields);

        // 将JavaRDD<String>转换为Row类型的RDD
        JavaRDD<Row> rowRDD = csvData.map(line -> {
            String[] parts = line.split(",");
            return RowFactory.create(parts[0], Integer.parseInt(parts[1]), Double.parseDouble(parts[2]));
        });

        // 使用定义好的Schema创建DataFrame
        Dataset<Row> df = spark.sqlContext().createDataFrame(rowRDD, schema);

        // 显示DataFrame的前几行数据
        df.show();

        // 关闭Spark上下文
        sc.stop();
    }
}