如何通過ServerlessSpark訪問OSS資料來源 - Data Lake Analytics - Deprecated

本文介紹如何通過Serverless Spark訪問OSS資料來源。您需要先配置訪問OSS的許可權，然後可以使用SQL的方式或者提交程式碼封裝（Python或者Jar包）的方式訪問OSS。

重要

雲原生資料湖分析（DLA）產品已退市，AnalyticDB for MySQL湖倉版支援DLA已有功能，並提供更多的功能和更好的效能。AnalyticDB for MySQL相關使用文檔，請參見訪問OSS資料來源。

操作步驟

配置DLA訪問OSS的許可權。
- 如果您使用的是阿里雲主帳號訪問OSS，則預設您擁有該帳號下所有OSS資料以及DLA OSS表的存取權限，無需配置，可直接使用。
- 如果您使用RAM子帳號訪問OSS並提交程式碼封裝作業，需要配置程式碼封裝和Spark代碼訪問OSS的許可權。具體操作請參見細粒度配置RAM子帳號許可權。
- 如果您使用Spark SQL訪問DLA OSS表的資料，需要確保您的RAM子帳號關聯了DLA帳號，並且DLA帳號擁有對應表的存取權限。如果您的RAM子帳號未關聯DLA帳號，請進行關聯操作，具體操作請參見DLA子帳號綁定RAM帳號。DLA帳號對於表的存取權限，您可以登入DLA控制台，在Serverless Presto > SQL執行頁面，使用GRANT或REVOKE文法進行操作。
配置Spark OSS Connector。
配置了OSS存取權限之後，您就可以使用Spark來訪問OSS資料了。在Spark的作業設定檔中，您需要添加配置項“spark.dla.connectors” : “oss”。DLA平台內建了Spark OSS Connector相關的實現，預設不生效，需要配置該參數令其生效。如果您有Spark OSS Connector的其他實現方式，您不需要配置該參數，您只需提交您自己的實現Jar包，並添加相應的配置即可。

訪問OSS資料。

您可以通過以下兩種方式訪問OSS資料：

通過提交Spark SQL語句的方式來訪問OSS資料，具體操作請參見Spark SQL。作業樣本配置如下所示：

{
    "sqls": [
        "select * from `1k_tables`.`table0` limit 100",
        "insert into `1k_tables`.`table0` values(1, 'test')"
    ],
    "name": "sql oss test",
    "conf": {
        "spark.dla.connectors": "oss",
        "spark.driver.resourceSpec": "small",
        "spark.sql.hive.metastore.version": "dla",
        "spark.executor.instances": 10,
        "spark.dla.job.log.oss.uri": "oss://test/spark-logs",
        "spark.executor.resourceSpec": "small"
    }
}

通過Java、Scala、Python代碼訪問OSS資料。下面以Scala為例進行說明：

{  
  "args": ["oss://${oss-buck-name}/data/test/test.csv"],
  "name": "spark-oss-test",
  "file": "oss://${oss-buck-name}/jars/test/spark-examples-0.0.1-SNAPSHOT.jar",
  "className": "com.aliyun.spark.oss.SparkReadOss",
  "conf": {
    "spark.driver.resourceSpec": "medium",
    "spark.executor.resourceSpec": "medium",
    "spark.executor.instances": 2,
    "spark.dla.connectors": "oss"
  }
}

說明

SparkReadOss對應的源碼可以參考DLA Spark OSS demo。

啟用OSS資料寫入效能最佳化

當您使用自建HiveMetaStore或者DLA中繼資料服務訪問OSS時，社區版Spark HiveClient的rename操作比較低效，DLA對此進行了最佳化。您只需要將參數spark.sql.hive.dla.metastoreV2.enable設定為true即可啟用這項最佳化。樣本如下：

{
  "args": ["oss://${oss-buck-name}/data/test/test.csv"],
  "name": "spark-oss-test",
  "file": "oss://${oss-buck-name}/jars/test/spark-examples-0.0.1-SNAPSHOT.jar",
  "className": "com.aliyun.spark.oss.WriteParquetFile",
  "conf": {
    "spark.driver.resourceSpec": "medium",
    "spark.executor.resourceSpec": "medium",
    "spark.executor.instances": 2,
    "spark.dla.connectors": "oss",
    "spark.sql.hive.dla.metastoreV2.enable": "true"
  }
}

OSS Connector資料寫入效能最佳化

OSS Connector資料寫入效能最佳化功能是DLA Spark團隊基於OSS分區上傳功能，針對Spark寫入資料到OSS過程中大量調用OSS API導致寫入效能差的問題，實現的效能最佳化提升。在典型情境下，效能可提升1~3倍。

您需要啟用DLA Spark內建的OSS connector，並開啟效能最佳化開關，才能使用該功能。具體配置如下：

spark.dla.connectors = oss;  //啟用DLA Spark內建的OSS connector。
spark.hadoop.job.oss.fileoutputcommitter.enable = true;  //開啟效能最佳化開關。

說明

如果啟用該效能最佳化功能，在作業被強制Kill等情況下，可能會產生一些沒有被清理的檔案片段，佔用您OSS的儲存空間。建議對相關OSS Bucket設定片段生命週期規則，對到期未合并的片段自動進行清理，建議配置周期為3天以上。具體操作請參見設定生命週期規則。
該效能最佳化功能對RDD的saveAsHadoop首碼和saveAsNewAPIHadoop首碼的方法不生效。

使用樣本：

{
  "args": ["oss://${oss-buck-name}/data/test/test.csv"],
  "name": "spark-oss-test",
  "file": "oss://${oss-buck-name}/jars/test/spark-examples-0.0.1-SNAPSHOT.jar",
  "className": "com.aliyun.spark.oss.WriteParquetFile",
  "conf": {
    "spark.driver.resourceSpec": "medium",
    "spark.executor.resourceSpec": "medium",
    "spark.executor.instances": 2,
    "spark.dla.connectors": "oss",
    "spark.hadoop.job.oss.fileoutputcommitter.enable": true
  }
}