Spark 1.6.3 Java版Spark Context编写及本地文件转DataFrame求助

首先看你提供的代码，里面混了Scala的语法（比如val关键字），而且缺少了将RDD转换为DataFrame的关键步骤，另外本地文件读取还需要注意Spark运行模式的配置，我给你整理了完整的Java实现方案：

核心要点说明 #

• 必须配置Spark运行在本地模式：因为文件存储在本地机器，需要通过setMaster("local[*]")指定，否则默认集群模式下Spark会去HDFS查找文件，必然失败。
• 要将RDD<String>转换为DataFrame，需要先定义数据Schema，再把文本行映射成Row类型的RDD，最后通过SQLContext创建结构化的DataFrame。

完整代码示例 #

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
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 LocalFileToDataFrame {
    public static void main(String[] args) {
        // 1. 配置Spark环境，指定本地模式和应用名称
        SparkConf sparkConf = new SparkConf()
                .setAppName("LocalFileToDataFrameSample")
                .setMaster("local[*]"); // 本地模式，自动使用所有可用CPU核心

        // 2. 初始化SparkContext和SQLContext
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
        SQLContext sqlContext = new SQLContext(sc);

        // 3. 读取本地文件：建议用绝对路径（比如"/Users/xxx/Documents/file1.txt"），相对路径要确保程序运行目录下有文件
        JavaRDD<String> file1Rdd = sc.textFile("file1.txt");
        JavaRDD<String> file2Rdd = sc.textFile("file2.txt");

        // 4. 定义DataFrame的Schema（这里假设每行是单个文本字段，命名为content）
        List<StructField> fields = new ArrayList<>();
        fields.add(DataTypes.createStructField("content", DataTypes.StringType, true));
        StructType schema = DataTypes.createStructType(fields);

        // 5. 将文本RDD转换为Row类型的RDD
        JavaRDD<Row> file1RowRdd = file1Rdd.map(line -> RowFactory.create(line));
        JavaRDD<Row> file2RowRdd = file2Rdd.map(line -> RowFactory.create(line));

        // 6. 创建DataFrame
        DataFrame file1Df = sqlContext.createDataFrame(file1RowRdd, schema);
        DataFrame file2Df = sqlContext.createDataFrame(file2RowRdd, schema);

        // 7. 验证结果：打印Schema和前10行数据
        file1Df.printSchema();
        file1Df.show(10);

        file2Df.printSchema();
        file2Df.show(10);

        // 关闭资源
        sc.stop();
    }
}

额外注意事项 #

• 如果你的文本是结构化格式（比如CSV，每行用逗号分隔多字段），可以修改映射逻辑，把每行拆分后创建多字段的Row，同时更新Schema即可。
• 依赖配置：如果用Maven管理项目，需要在pom.xml中添加Spark核心和SQL的依赖：

<dependencies>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.10</artifactId>
        <version>1.6.3</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.10</artifactId>
        <version>1.6.3</version>
    </dependency>
</dependencies>

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