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 形式進行數據的讀取。該功能實現的類圖展現以下：

1. 定義數據源 NebulaRelatioProvider，繼承 RelationProvider 進行 relation 自定義，繼承 DataSourceRegister 進行外部數據源的註冊。
2. 定義 NebulaRelation 定義 Nebula Graph 的數據 Schema 和數據轉換方法。在 getSchema() 方法中鏈接 Nebula Graph 的 Meta 服務獲取配置的返回字段對應的 Schema 信息。
3. 定義 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 git@github.com:vesoft-inc/nebula-java.git
cd nebula-java/tools/nebula-spark
mvn clean compile package install -Dgpg.skip -Dmaven.javadoc.skip=true

將編譯打成的包 copy 到本地 Maven 庫。

應用示例以下：

1. 在 mvn 項目的 pom 文件中加入 nebula-spark 依賴

<dependency>
  <groupId>com.vesoft</groupId>
  <artifactId>nebula-spark</artifactId>
  <version>1.1.0</version>
</dependency>

2. 在 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 代碼：https://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 拉你進交流羣~~