Spark Connector Reader 原理與實踐
本文主要講述如何利用 Spark Connector 進行 Nebula Graph 數據的讀取。
Spark Connector 簡介
Spark Connector 是一個 Spark 的數據連接器，可以通過該連接器進行外部數據系統的讀寫操作，Spark Connector 包含兩部分，分別是 Reader 和 Writer，而本文側重介紹 Spark Connector Reader，Writer 部分將在下篇和大家詳聊。
Spark Connector Reader 原理
Spark Connector Reader 是將 Nebula Graph 作為 Spark 的擴展數據源，從 Nebula Graph 中將數據讀成 DataFrame，再進行後續的 map、reduce 等操作。
Spark SQL 允許用戶自定義數據源，支援對外部數據源進行擴展。通過 Spark SQL 讀取的數據格式是以命名列方式組織的分散式數據集 DataFrame，Spark SQL 本身也提供了眾多 API 方便用戶對 DataFrame 進行計算和轉換，能對多種數據源使用 DataFrame 介面。
Spark 調用外部數據源包的是 org.apache.spark.sql，首先了解下 Spark SQL 提供的的擴展數據源相關的介面。
Basic Interfaces
- BaseRelation：表示具有已知 Schema 的元組集合。所有繼承 BaseRelation 的子類都必須生成 StructType 格式的 Schema。換句話說，BaseRelation 定義了從數據源中讀取的數據在 Spark SQL 的 DataFrame 中存儲的數據格式的。
- RelationProvider：獲取參數列表，根據給定的參數返回一個新的 BaseRelation。
- DataSourceRegister：註冊數據源的簡寫，在使用數據源時不用寫數據源的全限定類名，而只需要寫自定義的 shortName 即可。
Providers
- RelationProvider：從指定數據源中生成自定義的 relation。
createRelation()會基於給定的 Params 參數生成新的 relation。 - SchemaRelationProvider：可以基於給定的 Params 參數和給定的 Schema 資訊生成新的 Relation。
RDD
- RDD[InternalRow]: 從數據源中 Scan 出來後需要構造成 RDD[Row]
要實現自定義 Spark 外部數據源，需要根據數據源自定義上述部分方法。
在 Nebula Graph 的 Spark Connector 中，我們實現了將 Nebula Graph 作為 Spark SQL 的外部數據源，通過 sparkSession.read 形式進行數據的讀取。該功能實現的類圖展示如下：
- 定義數據源 NebulaRelatioProvider，繼承 RelationProvider 進行 relation 自定義，繼承 DataSourceRegister 進行外部數據源的註冊。
- 定義 NebulaRelation 定義 Nebula Graph 的數據 Schema 和數據轉換方法。在
getSchema()方法中連接 Nebula Graph 的 Meta 服務獲取配置的返回欄位對應的 Schema 資訊。 - 定義 NebulaRDD 進行 Nebula Graph 數據的讀取。
compute()方法中定義如何讀取 Nebula Graph 數據，主要涉及到進行 Nebula Graph 數據 Scan、將讀到的 Nebula Graph Row 數據轉換為 Spark 的 InternalRow 數據，以 InternalRow 組成 RDD 的一行，其中每一個 InternalRow 表示 Nebula Graph 中的一行數據，最終通過分區迭代的形式將 Nebula Graph 所有數據讀出組裝成最終的 DataFrame 結果數據。
Spark Connector Reader 實踐
Spark Connector 的 Reader 功能提供了一個介面供用戶編程進行數據讀取。一次讀取一個點/邊類型的數據，讀取結果為 DataFrame。
下面開始實踐，拉取 GitHub 上 Spark Connector 程式碼：
git clone -b v1.0 [email protected]:vesoft-inc/nebula-java.git
cd nebula-java/tools/nebula-spark
mvn clean compile package install -Dgpg.skip -Dmaven.javadoc.skip=true
將編譯打成的包 copy 到本地 Maven 庫。
應用示例如下：
- 在 mvn 項目的 pom 文件中加入
nebula-spark依賴
<dependency>
<groupId>com.vesoft</groupId>
<artifactId>nebula-spark</artifactId>
<version>1.1.0</version>
</dependency>
- 在 Spark 程式中讀取 Nebula Graph 數據：
// 讀取 Nebula Graph 點數據
val vertexDataset: Dataset[Row] =
spark.read
.nebula("127.0.0.1:45500", "spaceName", "100")
.loadVerticesToDF("tag", "field1,field2")
vertexDataset.show()
// 讀取 Nebula Graph 邊數據
val edgeDataset: Dataset[Row] =
spark.read
.nebula("127.0.0.1:45500", "spaceName", "100")
.loadEdgesToDF("edge", "*")
edgeDataset.show()
配置說明：
- nebula(address: String, space: String, partitionNum: String)
address：可以配置多個地址，以英文逗號分割，如「ip1:45500,ip2:45500」
space: Nebula Graph 的 graphSpace
partitionNum： 設定spark讀取Nebula時的partition數，盡量使用創建 Space 時指定的 Nebula Graph 中的 partitionNum，可確保一個Spark的partition讀取Nebula Graph一個part的數據。
- loadVertices(tag: String, fields: String)
tag：Nebula Graph 中點的 Tag
fields：該 Tag 中的欄位，，多欄位名以英文逗號分隔。表示只讀取 fields 中的欄位，* 表示讀取全部欄位
- loadEdges(edge: String, fields: String)
edge：Nebula Graph 中邊的 Edge
fields：該 Edge 中的欄位，多欄位名以英文逗號分隔。表示只讀取 fields 中的欄位，* 表示讀取全部欄位
其他
Spark Connector Reader 的 GitHub 程式碼：//github.com/vesoft-inc/nebula-java/tree/master/tools/nebula-spark
在此特別感謝半雲科技所貢獻的 Spark Connector 的 Java 版本
參考資料
[1] Extending Spark Datasource API: write a custom spark datasource
[2] spark external datasource source code
喜歡這篇文章？來來來，給我們的 GitHub 點個 star 表鼓勵啦~~ 🙇♂️🙇♀️ [手動跪謝]
交流圖資料庫技術？交個朋友，Nebula Graph 官方小助手微信：NebulaGraphbot 拉你進交流群~~
