版權聲明：本文為王小雷原創文章，未經部落客允許不得轉載 https://blog.csdn.net/dream_an/article/details/81058073

14.4 Spark-SQL基于Cassandra資料分析程式設計執行個體

本節主要内容：

Spark對Canssandra資料庫資料的處理，通過Spark SQL對結構化資料進行資料分析。

• 建立Gradle項目，引入依賴
• 建立Spark Session連接配接
• 寫入Cassandra資料庫
• 讀取Cassandra資料庫
• Spark注冊SQL 臨時視圖執行Distinct操作

完整源碼Spark2Cassandra.java

完整項目源碼

14.3.1 引入依賴

compile group: 'org.apache.spark', name: 'spark-core_2.11', version: '2.3.1'
compile group: 'org.apache.spark', name: 'spark-sql_2.11', version: '2.3.1'
compile group: 'com.datastax.spark', name: 'spark-cassandra-connector_2.11', version: '2.3.0'           

複制

14.3.2 建立Spark Session

SparkSession spark = SparkSession
        .builder()
        .appName("Java Spark SQL basic example")
//            .config("spark.some.config.option", "some-value")
        .config("spark.cassandra.connection.host", "192.168.56.110")
        .config("spark.cassandra.auth.username", "busuanzi")
        .config("spark.cassandra.auth.password", "busuanzi.org")
        .config("spark.cassandra.connection.port", "9042")
        .getOrCreate();           

複制

14.3.3 寫入Cassandra資料庫表

ds1.write()
        .format("org.apache.spark.sql.cassandra")
        .options(new HashMap<String, String>() {
            {
                put("keyspace", "busuanzi_org");
                put("table", "top_n_url");
            }
        }).mode("append").save();           

複制

查詢Canssandra資料庫

14.3.4 讀取Cassandra資料庫表

Dataset<Row> ds = spark.read()
        .format("org.apache.spark.sql.cassandra")
        .options(new HashMap<String, String>(){
            {
                put("keyspace", "busuanzi_org");
                put("table", "top_n_url");
            }
        }).load();
ds.show();           

複制

14.3.5 Spark SQL Distinct去重

将ds DataFrame注冊為SQL臨時視圖

ds.createOrReplaceTempView("dsv");           

複制

Spark SQL去重

ds.select("username", "projects", "comment").distinct().show();           

複制

完整源碼Spark2Cassandra.java

完整項目源碼

14.3.6 在IDEA運作項目設定

源碼擷取 https://github.com/wangxiaoleiAI/big-data