Spark 鏈接mysql 及MongoDB

時間 2019-12-01

原文原文鏈接

在spark 運算過程當中，經常須要鏈接不一樣類型的數據庫以獲取或者存儲數據，這裏將說起Spark如何鏈接mysql和MongoDB.java

1. 鏈接mysql , 在1.3版本提出了一個新概念DataFrame ，所以如下方式獲取到的是DataFrame，可是可經過JavaRDD<Row> rows = jdbcDF.toJavaRDD()轉化爲JavaRDD。mysql

import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

public class Main implements Serializable {

    private static final org.apache.log4j.Logger LOGGER = org.apache.log4j.Logger.getLogger(Main.class);

    private static final String MYSQL_DRIVER = "com.mysql.jdbc.Driver";
    private static final String MYSQL_USERNAME = "expertuser";
    private static final String MYSQL_PWD = "expertuser123";
    private static final String MYSQL_CONNECTION_URL =
            "jdbc:mysql://localhost:3306/employees?user=" + MYSQL_USERNAME + "&password=" + MYSQL_PWD;

    private static final JavaSparkContext sc =
            new JavaSparkContext(new SparkConf().setAppName("SparkJdbcDs").setMaster("local[*]"));

    private static final SQLContext sqlContext = new SQLContext(sc);

    public static void main(String[] args) {
        //Data source options
        Map<String, String> options = new HashMap<>();
        options.put("driver", MYSQL_DRIVER);
        options.put("url", MYSQL_CONNECTION_URL); //getConnection 返回一個已經打開的結構化數據庫鏈接，JdbcRDD會自動維護關閉。
        options.put("dbtable",
                    "(select emp_no, concat_ws(' ', first_name, last_name) as full_name from employees) as employees_name");
//　　　　 sql 是查詢語句，此查詢語句必須包含兩處佔位符?來做爲分割數據庫ResulSet的參數，例如："select title, author from books where ? < = id and id <= ?"
        options.put("partitionColumn", "emp_no");//進行分區的表字段
        options.put("lowerBound", "10001");
//　　　　 owerBound, upperBound, numPartitions 分別爲第1、第二佔位符，partition的個數。例如，給出lowebound 1，upperbound 20， numpartitions 2，則查詢分別爲(1, 10)與(11, 20)
        options.put("upperBound", "499999");
        options.put("numPartitions", "10");

        //Load MySQL query result as DataFrame
        DataFrame jdbcDF = sqlContext.load("jdbc", options);
        JavaRDD<Row> rows = jdbcDF.toJavaRDD(); 

        List<Row> employeeFullNameRows = jdbcDF.collectAsList();

        for (Row employeeFullNameRow : employeeFullNameRows) {
            LOGGER.info(employeeFullNameRow);
        }
    }
}