033 Java Spark的編程

1.Java SparkCore編程

　　入口是：JavaSparkContext
　　　　基本的RDD是：JavaRDD
　　　　其他常用RDD: JavaPairRDD
　　JavaRDD和JavaPairRDD轉換：
　　　　JavaRDD => JavaPairRDD: 通過mapToPair函數
　　　　JavaPairRDD => JavaRDD: 通過map函數轉換

 

2.前提

　　運行前將core-site.xml復制到resource文件夾中

 

3.程序　　

package com.ibeifeng.senior;

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.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Arrays;
import java.util.Iterator;

/**
 * Java實現Spark的WordCount程序
 * Created by ibf on 02/15.
 */
public class JavaWordCountSparkCore {
    public static void main(String[] args) {
        String resultHDFSSavePath = "/beifeng/spark/result/wordcount/" + System.currentTimeMillis();
        // 1. 創建SparkConf配置信息
        SparkConf conf = new SparkConf()
                .setMaster("local[*]")
                .setAppName("spark-wordcount");

        // 2. 創建SparkContext對象，在java編程中，該對象叫做JavaSparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);

        // 3. 從hdfs讀取文件形成RDD
        // TODO: 文件路徑自行給定
        JavaRDD<String> rdd = sc.textFile("/hive/common.db/dept");

        // 4. RDD數據處理
        // TODO: 過濾特殊字符
        // 4.1 行數據的分割，調用flatMap函數
        JavaRDD<String> wordRDD = rdd.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterable<String> call(String s) throws Exception {
                String line = s;
                if (line == null) line = "";
                String[] arr = line.split("\t");
                return Arrays.asList(arr);
            }
        });

        // 4.2 將數據轉換為key/value鍵值對
        /**
         * RDD的reduceByKey函數不是RDD類中，通過隱式轉換后，存在於其他類中<br/>
         * Java由於不存在隱式轉換，所以不能直接調用map函數進行key/value鍵值對轉換操作，必須調用特定的函數
         * */
        JavaPairRDD<String, Integer> wordCountRDD = wordRDD.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<String, Integer>(s, 1);
            }
        });

        // 4.3 聚合結果
        JavaPairRDD<String, Integer> resultRDD = wordCountRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {

            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });

        // 5. 結果輸出
        // 5.1 結果輸出到HDFS
        resultRDD.saveAsTextFile(resultHDFSSavePath);
        // 5.2 結果輸出到MySQL
        /**
         * SparkCore RDD數據的讀入是通過InputFormat來讀入數據形成RDD的
         *  sc.newAPIHadoopRDD[K, V, F <: NewInputFormat[K, V]](
         conf: Configuration,
         fClass: Class[F],
         kClass: Class[K],
         vClass: Class[V])
         * RDD的saveASxxxx相關方法是利用OutputFormat來進行數據輸出的
         * resultRDD.saveAsNewAPIHadoopDataset(conf: Configuration);
         */
        resultRDD.foreachPartition(new VoidFunction<java.util.Iterator<Tuple2<String, Integer>>>() {

            @Override
            public void call(Iterator<Tuple2<String, Integer>> tuple2Iterator) throws Exception {
                Class.forName("com.mysql.jdbc.Driver");
                String url = "jdbc:mysql://hadoop-senior01:3306/test";
                String username = "root";
                String password = "123456";
                Connection conn = null;
                try {
                    // 1. 創建connection連接
                    conn = DriverManager.getConnection(url, username, password);

                    // 2. 構建statement
                    String sql = "insert into wordcount values(?,?)";
                    PreparedStatement pstmt = conn.prepareStatement(sql);

                    // 3. 結果數據輸出
                    while (tuple2Iterator.hasNext()) {
                        Tuple2<String, Integer> t2 = tuple2Iterator.next();
                        pstmt.setString(1, t2._1());
                        pstmt.setLong(2, t2._2());

                        pstmt.executeUpdate();
                    }
                } finally {
                    // 4. 關閉連接
                    conn.close();
                }

            }
        });


    }
}