Spark Java創建DataFrame

俗話說得好，磨刀不誤砍柴工，獻上一副來自國家5A級風景區美景圖。

述說正傳，接下來開始說正事。

以前用Python和Scala操作Spark的時候比較多，畢竟Python和Scala代碼寫起來要簡潔很多。

今天一起來看看Java版本怎么創建DataFrame，代碼寫起來其實差不多，畢竟公用同一套API。測試數據可以參考我之前的文章。

先來總結下Spark的一般流程：

1，先創建Spark基礎變量，spark，sc

2，加載數據，rdd.textFile，spark.read.csv/json等

3，數據處理，mapPartition, map，filter，reduce等一系列transformation操作

4，數據保存，saveAstextFile，或者其他DataFrame方法

祭出代碼

package dev.java;

import dev.utils.Utils;
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.SparkSession;
import org.apache.spark.sql.types.StructType;
import scala.Tuple2;

import java.util.List;

public class Spark1 {

    private static final String fileData = "seed";
    private static final String fileSave = "result";
    private static SparkSession spark = SparkSession.builder()
                .appName("Java-Spark")
                .master("local[*]")
                .config("spark.default.parallelism", 100)
                .config("spark.sql.shuffle.partitions", 100)
                .config("spark.driver.maxResultSize", "3g")
                .getOrCreate();
    private static JavaSparkContext sc = JavaSparkContext.fromSparkContext(spark.sparkContext());

    public static void main(String[] args) {
        Utils.delete(fileSave);
        //
        t1();
    }

    private static void t1() {
        JavaRDD<Row> rdd = sc.textFile(fileData)
                .map(v -> {
                    String[] parts = v.split("\t");
                    return RowFactory.create(parts[0], Long.parseLong(parts[1]));
                })
                .filter(v -> v.getLong(1) >= 10000)
                .sortBy(v -> v.getLong(1), false, 100)
                .coalesce(2);
        Dataset<Row> df = spark.createDataFrame(rdd, StructType.fromDDL("title string, qty long"));
        df.write().csv(fileSave);
        spark.stop();
    }
}