Spark對接OSS提供了高效、靈活的數據處理和分析方式,將大數據處理和云存儲相結合。本文介紹Spark如何處理和分析OSS中的數據。
背景信息
當前E-MapReduce:
支持MetaService服務。
支持通過免AccessKey方式訪問OSS數據源。
支持通過顯式寫AccessKey和Endpoint方式訪問OSS數據源。
說明OSS Endpoint需使用內網域名。域名詳情信息,請參見OSS Endpoint。
Spark RDD接入OSS示例
本示例為您展示,Spark如何以免AccessKey方式讀取OSS中數據,并將處理完的數據寫回至OSS。
在spark-shell
中執行如下代碼。
val conf = new SparkConf().setAppName("Test OSS")
val sc = new SparkContext(conf)
val pathIn = "oss://bucket/path/to/read"
val inputData = sc.textFile(pathIn)
val cnt = inputData.count
println(s"count: $cnt")
val outputPath = "oss://bucket/path/to/write"
val outpuData = inputData.map(e => s"$e has been processed.")
outpuData.saveAsTextFile(outputPath)
完整示例代碼,請參見Spark對接OSS。
本文示例中以下參數請根據您的實際情況替換:
pathIn
:定義了要讀取的文件路徑。outputPath
:定義了要寫入的文件路徑。
PySpark接入OSS示例
本示例為您展示,PySpark如何以免AccessKey方式讀取OSS中數據,并將處理完的數據寫回至OSS。
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("Python Spark SQL OSS example").getOrCreate()
pathIn = "oss://bucket/path/to/read"
df = spark.read.text(pathIn)
cnt = df.count()
print(cnt)
outputPath = "oss://bucket/path/to/write"
df.write.format("parquet").mode('overwrite').save(outputPath)
Spark-SQL創建CSV表示例
本示例為您展示Spark-SQL如何創建CSV表,并存放至OSS,操作步驟如下。
在
spark-sql
中執行如下命令。CREATE DATABASE test_db LOCATION "oss://test_bucket/test_db"; USE test_db; CREATE TABLE student (id INT, name STRING, age INT) USING CSV options ("delimiter"=";", "header"="true"); INSERT INTO student VALUES(1, "ab", 12); SELECT * FROM student;
執行完以上命令,返回以下信息。
1 ab 12
命令中涉及參數如下:
test_bucket
:OSS Bucket的名稱。delimiter
:指定CSV文件中數據的分隔符。header
:指定CSV文件中第一行是否是表頭,可設置true
表示是,false
表示否。
查看存儲在OSS上的CSV文件。
CSV文件的第一行包含表頭,并使用分號(;)作為字段的分隔符。示例內容如下。
id;name;age 1;ab;12
集群顯示指定訪問OSS的AccessKey
本示例將為您展示集群如何顯示指定訪問OSS的AccessKey,操作步驟如下。
更改免密配置。
EMR集群訪問OSS是免密配置。您可以通過HDFS組件core-site.xml中的fs.oss.credentials.provider配置項控制免密訪問。
如果想去掉免密,使用AccessKey訪問,需要把fs.oss.credentials.provider參數去掉,同時在core-site.xml中加上下面AccessKey的參數。
<property> <name>fs.oss.accessKeyId</name> <value>LTAI5tM85Z4sc****</value> </property> <property> <name>fs.oss.accessKeySecret</name> <value>HF7P1L8PS6Eqf****</value> </property>
執行以下命令驗證是否生效。
hadoop fs -ls oss://test_bucket/test_db
返回以下信息,發現無法訪問OSS。
ls: ERROR: without login secrets configured.
加上AccessKey配置之后,重新執行命令
hadoop fs -ls oss://test_bucket/test_db
,則能夠查看OSS文件目錄。drwxrwxrwx - root root 0 2022-11-30 12:51 oss://test_bucket/test_db/student
重啟Spark相關服務,查看Spark相關服務是否正常。