實戰 | 將Apache Hudi資料集寫入阿里雲OSS
阿新 • • 發佈:2020-04-25
### 1. 引入
雲上物件儲存的廉價讓不少公司將其作為主要的儲存方案,而Hudi作為資料湖解決方案,支援物件儲存也是必不可少。之前AWS EMR已經內建整合Hudi,也意味著可以在S3上無縫使用Hudi。當然國內使用者可能更多使用阿里雲OSS作為雲上儲存方案,那麼如果使用者想基於OSS構建資料湖,那麼Hudi是否支援呢?隨著Hudi社群主分支已經合併了支援OSS的PR,現在只需要基於master分支build版本即可,或者等待下一個版本釋出便可直接使用,經過簡單的配置便可將資料寫入OSS。
### 2. 配置
#### 2.1 pom依賴
需要額外新增的主要pom依賴如下
```xml
org.apache.hadoop
hadoop-aliyun
3.2.1
com.aliyun.oss
aliyun-sdk-oss
3.8.1
```
#### 2.2 core-site.xml配置
若需訪問OSS,需要修改core-site.xml,關鍵配置如下
```xml
fs.defaultFS
oss://bucketname/
fs.oss.endpoint
oss-endpoint-address
Aliyun OSS endpoint to connect to.
fs.oss.accessKeyId
oss_key
Aliyun access key ID
fs.oss.accessKeySecret
oss-secret
Aliyun access key secret
fs.oss.impl
org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem
```
### 3. 原始碼
示例原始碼如下
```java
import org.apache.hudi.QuickstartUtils.*;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import java.io.IOException;
import java.util.List;
import static org.apache.hudi.QuickstartUtils.convertToStringList;
import static org.apache.hudi.QuickstartUtils.getQuickstartWriteConfigs;
import static org.apache.hudi.config.HoodieWriteConfig.TABLE_NAME;
import static org.apache.spark.sql.SaveMode.Overwrite;
public class OssHudiDemo {
public static void main(String[] args) throws IOException {
SparkSession spark = SparkSession.builder().appName("Hoodie Datasource test")
.master("local[2]")
.config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
.config("spark.io.compression.codec", "snappy")
.config("spark.sql.hive.convertMetastoreParquet", "false")
.getOrCreate();
JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
String tableName = "hudi_trips_cow";
String basePath = "/tmp/hudi_trips_cow";
DataGenerator dataGen = new DataGenerator();
List inserts = convertToStringList(dataGen.generateInserts(10));
Dataset df = spark.read().json(jsc.parallelize(inserts, 2));
df.write().format("org.apache.hudi").
options(getQuickstartWriteConfigs()).
option(TABLE_NAME, tableName).
mode(Overwrite).
save(basePath);
Dataset roViewDF = spark.read().format("org.apache.hudi").load(basePath + "/*/*/*");
roViewDF.registerTempTable("hudi_ro_table");
spark.sql("select * from hudi_ro_table").show(false);
spark.stop();
}
}
```
即先寫入OSS,下圖可以看到OSS的Bucket中已經成功寫入了資料,然後再通過spark查詢寫入的結果。
![](https://img2020.cnblogs.com/blog/616953/202004/616953-20200425145746705-629205626.png)
部分查詢結果如下
```xml
|20200421205942 |20200421205942_2_10 |6fd496f8-ebee-4f67-8f86-783ff3fed3ab|asia/india/chennai |1f71bed9-833b-4fca-8b4b-4cd014bdf88a-0_2-22-30_20200421205942.parquet|0.40613510977307 |0.5644092139040959 |driver-213|0.798706304941517 |0.02698359227182834|17.851135255091155|asia/india/chennai |rider-213|0.0|6fd496f8-ebee-4f67-8f86-783ff3fed3ab|
```
所有原始碼已經上傳至[https://github.com/leesf/oss-hudi-demo](https://github.com/leesf/oss-hudi-demo)
### 4. 最後
本篇文章很簡單,只用作展示如何通過Hudi將資料寫入OSS。當資料寫入OSS後,便可打通阿里雲上幾乎所有產品,這使得基於阿里雲技術棧進行資料湖分析將變得非常簡單,比如使用DLA(Data Lake Analytics),對標AWS的Athena,對Hudi資料集進行分析查詢,一體化的流程會讓分析變得異常