SparkRDD 转存 HIVE

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import org.apache.spark.sql.*;
import org.apache.spark.sql.hive.HiveContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import scala.Tuple2;

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

public class TaskResultSaveAndUpdate {
    public static void main(String[] args) {
        
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("hive");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SparkSession jsc = SparkSession.builder().appName("Test").master("local").enableHiveSupport().getOrCreate();
        HiveContext hiveCtx = new HiveContext(jsc);
        SQLContext sqlCtx = new SQLContext(jsc);
       
			<逻辑略>
        JavaRDD<Row> map = stringTuple2JavaPairRDD.map(f -> {
			<逻辑略>
            return RowFactory.create(user, A1, A2, A3, A4, A5, A6);
        });
        
        List<StructField> structFields = new ArrayList<StructField>();
        structFields.add(DataTypes.createStructField( "user", DataTypes.StringType, true ));
        structFields.add(DataTypes.createStructField( "A1", DataTypes.StringType, true ));
        structFields.add(DataTypes.createStructField( "A2", DataTypes.StringType, true ));
        structFields.add(DataTypes.createStructField( "A3", DataTypes.StringType, true ));
        structFields.add(DataTypes.createStructField( "A4", DataTypes.StringType, true ));
        structFields.add(DataTypes.createStructField( "A5", DataTypes.StringType, true ));
        structFields.add(DataTypes.createStructField( "A6", DataTypes.StringType, true ));
        StructType structType = DataTypes.createStructType(structFields);
        Dataset<Row> brandDF = sqlCtx.createDataFrame(map,structType);
        brandDF.registerTempTable("tableTmp");

        hiveCtx.sql("use test");
        hiveCtx.sql("CREATE TABLE IF NOT EXISTS tableHiveTmp (user string,A1 string,A2 string,A3 string,A4 string,A5 string, A6 string)");
        hiveCtx.sql("insert into tableHiveTmp select * from tableTmp");

    }
}