述說(shuō)正傳，接下來(lái)開(kāi)始說(shuō)正事。

以前用Python和Scala操作Spark的時(shí)候比較多，畢竟Python和Scala代碼寫(xiě)起來(lái)要簡(jiǎn)潔很多。

今天一起來(lái)看看Java版本怎么創(chuàng)建DataFrame，代碼寫(xiě)起來(lái)其實(shí)差不多，畢竟公用同一套API。測(cè)試數(shù)據(jù)可以參考我之前的文章。

先來(lái)總結(jié)下Spark的一般流程：

1，先創(chuàng)建Spark基礎(chǔ)變量，spark，sc

2，加載數(shù)據(jù)，rdd.textFile，spark.read.csv/json等

3，數(shù)據(jù)處理，mapPartition, map，filter，reduce等一系列transformation操作

4，數(shù)據(jù)保存，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(); }}

以上就是JAVA操作spark創(chuàng)建DataFrame的方法的詳細(xì)內(nèi)容，更多關(guān)于JAVA Spark 創(chuàng)建DataFrame的資料請(qǐng)關(guān)注好吧啦網(wǎng)其它相關(guān)文章！