保存为拼花文件在火花java

Question

我是Spark的新手。我尝试在本地模式(windows)下使用spark java将csv文件保存为parquet。我得到了这个错误。

原因：org.apache.spark.Spark异常：写入行时任务失败

我引用了其他线程并禁用了spark推测

set("spark.speculation "，" false ")

我还是会出错。我在csv中只使用了两个专栏进行测试。

输入：

portfolio_id;portfolio_code
1000042;CHNTESTPF04
1000042;CHNTESTPF04
1000042;CHNTESTPF04
1000042;CHNTESTPF04
1000042;CHNTESTPF04
1000042;CHNTESTPF04
1000042;CHNTESTPF04

我的代码:

JavaPairRDD<Integer, String> rowJavaRDD = pairRDD.mapToPair(new PairFunction<String, Integer, String>() {
    private Double[] splitStringtoDoubles(String s){
       String[] splitVals = s.split(";");
       Double[] vals = new Double[splitVals.length];
       for(int i= 0; i < splitVals.length; i++){
           vals[i] = Double.parseDouble(splitVals[i]);
       }
       return vals;
    }

    @Override
    public Tuple2<Integer, String> call(String arg0) throws Exception {
        // TODO Auto-generated method stub
        return null;
    }
});


SQLContext SQLContext;
SQLContext = new org.apache.spark.sql.SQLContext(sc);

Dataset<Row> fundDF = SQLContext.createDataFrame(rowJavaRDD.values(), funds.class);
fundDF.printSchema();

fundDF.write().parquet("C:/test");

请帮助我在这里错过的东西。

Answer 1

在这里回答，所以它从@Glenne Helles Sindholt告诉的未回答部分离开。抱歉延迟发布此消息

我通过在Tuple2()中添加函数的split来解决错误，如下所示:

    public void run(String t, String u)

    { 

     public Tuple2<String,String> call(String rec){ 
        String[] tokens = rec.split(";"); 
        String[] vals = new String[tokens.length]; 
        for(int i= 0; i < tokens.length; i++)
       { 
            vals[i] =tokens[i]; 
        } 

      return new Tuple2<String, String>(tokens[0], tokens[1]); 

     } 
   });

Answer 2

请找到我的Java/Sark代码，用于1）加载CSV indo Spark数据集2）将数据集保存到拼花

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.SparkSession;

SparkSession spark = SparkSession
                    .builder()
                    .appName("csv2parquet")
                    .config("spark.sql.warehouse.dir", "/file:/tmp")
                    .master("local")
                    .getOrCreate();

final String dir = "test.csv";


Dataset<Row> ds = spark.read().option("header", true).option("inferSchema", true).csv(dir);

final String parquetFile = "test.parquet";
final String codec = "parquet";

ds.write().format(codec).save(parquetFile);

spark.stop();

把这个加到你的绒球里

<dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-mapreduce-client-core</artifactId>
                <version>2.8.1</version>
</dependency>