2. 程式人生 > >Spark學習（陸）- Spark操作外部資料來源

Spark學習（陸）- Spark操作外部資料來源

阿新 • • 發佈：2018-12-09

文章目錄

產生背景

每個Spark應用程式從讀取資料開始，到儲存資料結束

  • 載入和儲存資料是不容易的
    比如大資料場景解析關係型資料庫需要用工具如sqoop將資料轉到hbase等。
  • 解析原始資料也不容易 ：text/json/parquet
  • 資料直接的轉換也麻煩
  • 資料集儲存在各個儲存系統中無法統一拉取和推送
    在這裡插入圖片描述

使用者的需要：
方便快速從不同的資料來源（json、parquet、rdbms），經過混合處理（json join parquet），再將處理結果以特定的格式（json、parquet）寫回到指定的系統（HDFS、S3）上去

Spark SQL 1.2 ==> 外部資料來源API

概念

External Data Source API

  • 一種整合各種外部資料的擴充套件方法
  • 可以使用各種格式和儲存系統讀寫DataFrame
  • Data Source API可以自動裁剪列和推送過濾器到源(謂詞下推):parquet/JDBO
  • Data Source API在Spark 1.2提出
    在這裡插入圖片描述

目標

  • 對於開發人員只需要構建針對外部資料來源的庫
    開發人員：是否需要把程式碼合併到spark中？？？？
    比如weibo資料只需要通過–jars傳入就行。
  • 對於使用人員通過DataFrames很容易載入和保持資料來源
    使用者
    讀：spark.read.format(format)
    format
    build-in:內建的 json parquet jdbc csv(2.0+後屬於內建)
    packages: 外部的比如微博的資料 並不是spark內建 https://spark-packages.org/
    寫：people.write.format(“parquet”).save(“path”)

操作Parquet檔案資料

parquet是無法直接檢視的；所以這裡就不提供資料了

  • spark.read.format(“parquet”).load(path)
  • df.write.format(“parquet”).save(path)
package com.kun.ExternalDataSource

import org.apache.spark.sql.SparkSession

/**
 * Parquet檔案操作
 */
object ParquetApp {

  def main(args: Array[String]) {

    val spark = SparkSession.builder().appName("SparkSessionApp")
      .master("local[2]").getOrCreate()


    /**
     * spark.read.format("parquet").load 這是標準寫法
     */
    val userDF = spark.read.format("parquet").load("file:///home/hadoop/app/spark-2.1.0-bin-2.6.0-cdh5.7.0/examples/src/main/resources/users.parquet")

    userDF.printSchema()
    userDF.show()

    userDF.select("name","favorite_color").show

    userDF.select("name","favorite_color").write.format("json").save("file:///home/hadoop/tmp/jsonout")

    spark.read.load("file:///home/hadoop/app/spark-2.1.0-bin-2.6.0-cdh5.7.0/examples/src/main/resources/users.parquet").show

    //會報錯，因為sparksql預設處理的format就是parquet
    spark.read.load("file:///home/hadoop/app/spark-2.1.0-bin-2.6.0-cdh5.7.0/examples/src/main/resources/people.json").show

    spark.read.format("parquet").option("path","file:///home/hadoop/app/spark-2.1.0-bin-2.6.0-cdh5.7.0/examples/src/main/resources/users.parquet").load().show
    spark.stop()
  }

}

啟動spark-shell來測試；速度快
在這裡插入圖片描述
載入parquet資料
在這裡插入圖片描述
資料型別
在這裡插入圖片描述
檢視所有列
在這裡插入圖片描述
只儲存前兩列
在這裡插入圖片描述
在這裡插入圖片描述
在這裡插入圖片描述

spark預設處理parquet資料
spark.read.load(“file:///home/hadoop/app/spark-2.1.0-bin-2.6.0-cdh5.7.0/examples/src/main/resources/people.json”).show
會報錯：

RuntimeException: file:/home/hadoop/app/spark-2.1.0-bin-2.6.0-cdh5.7.0/examples/src/main/resources/people.json is not a Parquet file

//原始碼中：
  val DEFAULT_DATA_SOURCE_NAME = SQLConfigBuilder("spark.sql.sources.default")
    .doc("The default data source to use in input/output.")
    .stringConf
    .createWithDefault("parquet")

利用spark中的sql來處理parquet

#注意USING的用法
CREATE TEMPORARY VIEW parquetTable
USING org.apache.spark.sql.parquet
OPTIONS (
  path "/home/hadoop/app/spark-2.1.0-bin-2.6.0-cdh5.7.0/examples/src/main/resources/users.parquet"
)

SELECT * FROM parquetTable

讀取parquet很多種寫法
在這裡插入圖片描述

操作Hive表資料

  • spark.table(tableName)
  • df.write.saveAsTable(tableName)

測試資料：
dept表

10 ACCOUNTING NEW YORK
20 RESEARCH DALLAS
30 SALES CHICAGO
40 OPERATIONS BOSTON

emp表

7369 SMITH CLERK 7902 1980-12-17 800.00 20
7499 ALLEN SALESMAN 7698 1981-2-20 1600.00 300.00 30
7521 WARD SALESMAN 7698 1981-2-22 1250.00 500.00 30
7566 JONES MANAGER 7839 1981-4-2 2975.00 20
7654 MARTIN SALESMAN 7698 1981-9-28 1250.00 1400.00 30
7698 BLAKE MANAGER 7839 1981-5-1 2850.00 30
7782 CLARK MANAGER 7839 1981-6-9 2450.00 10
7788 SCOTT ANALYST 7566 1987-4-19 3000.00 20
7839 KING PRESIDENT 1981-11-17 5000.00 10
7844 TURNER SALESMAN 7698 1981-9-8 1500.00 0.00 30
7876 ADAMS CLERK 7788 1987-5-23 1100.00 20
7900 JAMES CLERK 7698 1981-12-3 950.00 30
7902 FORD ANALYST 7566 1981-12-3 3000.00 20
7934 MILLER CLERK 7782 1982-1-23 1300.00 10
8888 HIVE PROGRAM 7839 1988-1-23 10300.00

前置條件：啟動spark-shell

檢視hive表：
在這裡插入圖片描述
遍歷emp
在這裡插入圖片描述
在這裡插入圖片描述

在這裡插入圖片描述

spark.sql("select deptno, count(1) as mount from emp where group by deptno").filter("deptno is not null").write.saveAsTable("hive_table_1")

//會報錯：
org.apache.spark.sql.AnalysisException: Attribute name "count(1)" contains invalid character(s) among " ,;{}()\n\t=". Please use alias to rename it.;

將資料寫入hive
在這裡插入圖片描述
注意：
spark.sqlContext.setConf(“spark.sql.shuffle.partitions”,“10”)配置的是分割槽的數量
在生產環境中一定要注意設定spark.sql.shuffle.partitions，預設是200
在這裡插入圖片描述
在這裡插入圖片描述

操作MySQL表資料

在這裡插入圖片描述

操作MySQL的資料方法一:

spark.read.format(“jdbc”).option(“url”, “jdbc:mysql://localhost:3306/hive”).option(“dbtable”, “hive.TBLS”).option(“user”, “root”).option(“password”, “root”).option(“driver”, “com.mysql.jdbc.Driver”).load()

不加option(“driver”, “com.mysql.jdbc.Driver”)會有錯誤：java.sql.SQLException: No suitable driver
在這裡插入圖片描述
可以看到結構和mysql裡的是一樣的
在這裡插入圖片描述
在這裡插入圖片描述
指定列輸出：
在這裡插入圖片描述

操作MySQL的資料方法二:

import java.util.Properties
val connectionProperties = new Properties()
connectionProperties.put("user", "root")
connectionProperties.put("password", "root")
connectionProperties.put("driver", "com.mysql.jdbc.Driver")

val jdbcDF2 = spark.read.jdbc("jdbc:mysql://localhost:3306", "hive.TBLS", connectionProperties)

在spark-shell裡測試
在這裡插入圖片描述
在這裡插入圖片描述
寫入MySQL

// Saving data to a JDBC source
jdbcDF.write
  .format("jdbc")
  .option("url", "jdbc:postgresql:dbserver")
  .option("dbtable", "schema.tablename")
  .option("user", "username")
  .option("password", "password")
  .save()

jdbcDF2.write
  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)

操作MySQL的資料方法三:

進入spark sql操作

CREATE TEMPORARY VIEW jdbcTable
USING org.apache.spark.sql.jdbc
OPTIONS (
  url "jdbc:mysql://localhost:3306",
  dbtable "hive.TBLS",
  user 'root',
  password 'root',
  driver 'com.mysql.jdbc.Driver'
)

在這裡插入圖片描述
show tables
在這裡插入圖片描述
select * from jdbctable;
在這裡插入圖片描述

Hive和MySQL綜合使用

關聯MySQL和Hive表資料關聯操作

外部資料來源綜合案例
在MySQL裡建立：

create database spark;
use spark;

CREATE TABLE DEPT(
DEPTNO int(2) PRIMARY KEY,
DNAME VARCHAR(14) ,
LOC VARCHAR(13) ) ;

INSERT INTO DEPT VALUES(10,'ACCOUNTING','NEW YORK');
INSERT INTO DEPT VALUES(20,'RESEARCH','DALLAS');
INSERT INTO DEPT VALUES(30,'SALES','CHICAGO');
INSERT INTO DEPT VALUES(40,'OPERATIONS','BOSTON');

進行下列兩表的jion
在這裡插入圖片描述
在這裡插入圖片描述

package com.kun.ExternalDataSource

import org.apache.spark.sql.SparkSession

/**
 * 使用外部資料來源綜合查詢Hive和MySQL的表資料
 */
object HiveMySQLApp {

  def main(args: Array[String]) {
    val spark = SparkSession.builder().appName("HiveMySQLApp")
      .master("local[2]").getOrCreate()

    // 載入Hive表資料
    val hiveDF = spark.table("emp")

    // 載入MySQL表資料
    val mysqlDF = spark.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306").option("dbtable", "spark.DEPT").option("user", "root").option("password", "root").option("driver", "com.mysql.jdbc.Driver").load()

    // JOIN
    val resultDF = hiveDF.join(mysqlDF, hiveDF.col("deptno") === mysqlDF.col("DEPTNO"))
    resultDF.show


    resultDF.select(hiveDF.col("empno"),hiveDF.col("ename"),
      mysqlDF.col("deptno"), mysqlDF.col("dname")).show

    spark.stop()
  }

}

spark-shell裡測試
在這裡插入圖片描述
在這裡插入圖片描述
在這裡插入圖片描述
在這裡插入圖片描述
同理json；parquet；csv等都是可以進行同樣操作的。

相關推薦

Spark學習（陸）- Spark操作外部資料來源

文章目錄 產生背景 概念 目標 操作Parquet檔案資料 操作Hive表資料 操作MySQL表資料 操作MySQL的資料方法一: 操作MySQL的資料方法二: 操作MySQL

Spark學習（拾）- Spark Streaming進階與案例實戰

實戰之updateStateByKey運算元的使用 updateStateByKey操作允許您在使用新資訊不斷更新狀態的同時維護任意狀態。要使用它，您需要執行兩個步驟。 1、定義狀態——狀態可以是任意資料型別。 2、定義狀態更新函式——用函式指定如何使用以前的狀態和輸入流中的新值更新

Spark學習（玖）- Spark Streaming核心概念與程式設計

文章目錄 核心概念之StreamingContext 核心概念之DStream 核心概念之Input DStreams和Receivers 基本資源 高階資源 核心概念之Transformat

Spark學習（捌）- Spark Streaming入門

文章目錄 spark概念 Spark Streaming應用場景 Spark Streaming整合Spark生態系統的使用 Spark Streaming發展史 從詞頻統計功能著手入門Spark Streaming

Spark學習（柒）- Spark SQL擴充套件和總結

文章目錄 Spark SQL使用場景 Spark SQL載入資料 1) RDD DataFrame/Dataset 2) Local Cloud(HDFS/S3) DataFrame與SQL的對比

Spark學習（一）--Spark入門介紹和安裝

本次主要介紹spark的入門概念和安裝 Spark概念 Spark安裝 Spark HA 高可用部署 1. Spark概念 1.1 什麼是Spark Spark 是一種快速、 通用、 可擴充套件的大資料分析引擎， 2009 年誕生於加州大學伯克利分校 AM

Spark學習（八）---Spark streaming原理

這次我們介紹spark streaming,今天主要是原理和相關的操作 Spark Streaming概念介紹 Spark Streaming的相關操作 1. Spark Streaming概念 1.1什麼是Spark Streaming Spark St

Spark學習（一）Spark介紹

一、什麼是spark spark是基於記憶體計算的大資料平行計算框架，也是hadoop中的mapreduce的替代方案，但和mapreduce又有許多不同。       Spark包含了大資料領域常見的各種計算框架：比如Spark Core用於離線計算，Spark SQL

spark中flatMap函數用法--spark學習（基礎）

比較 一次 ica 例子 tail details word fix spark spark中flatMap函數用法--spark學習（基礎） 在spark中map函數和flatMap函數是兩個比較常用的函數。其中 map：對集合中每個元素進行操作。 fl

spark學習（1）--ubuntu14.04集群搭建、配置（jdk）

RM int 5.0 java_home 輸入 str cas Go 比較 環境：ubuntu14.04 1、文本模式桌面模式切換 ctrl+alt+F6 切換到文本模式 ctrl + alt +F7 /輸入命令startx切換到桌面模式 2、更改Ip地址、主機名 /

Spark學習（二）——RDD的設計與運行原理

center data 創建 組成 分享圖片 img medium 列操作 信息 Spark的核心是建立在統一的抽象RDD之上，使得Spark的各個組件可以無縫進行集成，在同一個應用程序中完成大數據計算任務。RDD的設計理念源自AMP實驗室發表的論文《Resilient

Spark學習（伍）- DateFrame&Dataset

文章目錄 DataFrame產生背景 DataFrame概述 DataFrame和RDD的對比 DataFrame基本API操作 DataFrame與RDD互操作方式 DataFrame API操作案例實戰

Spark學習（肆）- 從Hive平滑過渡到Spark SQL

文章目錄 SQLContext的使用 HiveContext的使用 SparkSession的使用 spark-shell&spark-sql的使用 spark-shell spark-sql thri

Spark學習（叄）- 環境搭建

文章目錄 Spark原始碼編譯 1 寫在前面的話 2 前置準備 2.1 Java7的安裝 2.1.1 下載 2.1.2 安裝 2.2 Maven3.3.9的安裝

Spark 學習（6）

SparkStream 邏輯 當ssc啟動之後，driver會執行一個長時間執行的Task 作為Reveiver的executors，接受傳來的資料收到資料，並將其分成塊儲存在記憶體中 這寫塊也會被賦值給另一個Executors，以免資料丟失 每個

Spark學習（五）---RDD原理解析和spark執行架構

這次我們介紹RDD的原理和spark執行機制 RDD依賴關係 RDD快取 RDD容錯機制 spark執行架構 spark任務排程 1. RDD原理 首先我們對之前的單詞統計的程式碼做一個畫圖展示 1.1 RDD依賴關係 RDD和它依賴的父RDD的關係有兩

Spark學習（二）

RDD（resilient distributed dataset） RDD概念 RDD（Resilient Distributed Dataset）是一個彈性分散式資料集，是SPark提供的抽象的彈性分散式資料集(RDD)，它是可以並行操作的跨叢集節點的元素集

Spark學習（一）——Scala基礎學習

scala是一門多正規化程式語言，集成了面向物件程式設計和函數語言程式設計等多種特性。scala執行在虛擬機器上，併兼容現有的Java程式。Scala原始碼被編譯成java位元組碼，所以執行在JVM上，並可以呼叫現有的Java類庫。1、第一個Scala程式:Scala和Jav

Mysql學習（三）Spark（Scala）寫入Mysql的兩種方式

package total import org.apache.spark.sql.{DataFrame, Row, SQLContext, SparkSession} import org.apache.spark.{SparkConf, SparkContext} imp

Spark學習（一）—— 論文翻譯

================================ 彈性分散式資料集：基於記憶體叢集計算的容錯抽象 摘要        我們提出的彈性分散式資料集（RDDs），是一個讓程式設計師在大型叢集上以容錯的方式執行基於記憶體計算的分散式記憶體抽象