如何使用 Google Cloud Storage 作为 Delta Lake 的存储层？答案

【问题标题】：How to use Google Cloud Storage as a storage layer of Delta Lake?如何使用 Google Cloud Storage 作为 Delta Lake 的存储层？
【发布时间】：2020-01-04 14:24:13
【问题描述】：

我可以使用 Google Cloud Storage 作为 Delta Lake 的存储层吗？

发现于slack。

【问题讨论】：

标签： google-cloud-storage delta-lake

【解决方案1】：

Delta Lake 在 0.5.0 及以下版本中是不可能的。

在https://github.com/delta-io/delta/issues/294 跟踪此问题存在问题。随意投票以帮助优先考虑。

就在 Google 发布 Getting started with new table formats on Dataproc 一天后：

我们宣布，表格格式项目 Delta Lake 和 Apache Iceberg（孵化）现已在最新版本的 Cloud Dataproc（1.5 版预览版）中提供。您现在可以通过 Spark 或 Presto 开始使用它们。 Apache Hudi 也可用于 Dataproc 1.3。

【讨论】：

如果将 BigQuery 用作 BI/Reporting/Analytics 的服务渠道，就无法将 DeltaLake 表与 Bigquery 集成对吗？
@AshikaUmangaUmagiliya 你见过this吗？
此连接器用于从 Spark 应用读取/写入 BigQuery 数据。但我的要求不同。假设我们使用 delta-lake 格式来创建 Delta、Bronze 和 Gold 层（live-delta-io.pantheonsite.io/wp-content/uploads/2019/04/…）。然后 Gold 层的结果数据集需要通过 BigQuery 提供，这是不可能的，因为 BQ 无法读取 DeltaLake 文件
为什么不加载这个黄金 Delta 表并将其保存到 BigQuery？
是的，这是可能的。但这在黄金层和服务层之间增加了一个额外的“加载”层？如果我使用 Presto/StarBurst 作为服务层，则不需要这个额外的“加载”层。好吧..由于 BQ 是一项专有技术，因此我们不能指望“DeltaLake”格式支持将它们作为外部表加载到 BQ 中。

【解决方案2】：

这是可能的。这是示例代码和您需要的库：

确保首先设置您的凭据，您可以作为代码的一部分或作为环境：

export GOOGLE_APPLICATION_CREDENTIALS={gcs-key-path.json}

import org.apache.spark.sql.{SparkSession, DataFrame}
import com.google.cloud.spark.bigquery.repackaged.com.google.cloud.bigquery.BigQueryException
import com.google.cloud.spark.bigquery.repackaged.com.google.cloud.bigquery.BigQueryOptions
import com.google.cloud.spark.bigquery.repackaged.com.google.cloud.bigquery.DatasetInfo

spark.conf.set("parentProject", {Proj})
spark.conf.set("spark.hadoop.fs.gs.auth.service.account.enable", "true")   
spark.conf.set("spark.hadoop.fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem")
spark.conf.set("fs.AbstractFileSystem.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS")
spark.conf.set("spark.delta.logStore.gs.impl", "io.delta.storage.GCSLogStore")
spark.conf.set("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")


val targetTablePath = "gs://{bucket}/{dataset}/{tablename}"
spark.range(5, 10).write.format("delta")
      .mode("overwrite")
      .save(targetTablePath)

您需要的库：

"io.delta" % "delta-core_2.12" % "1.0.0",
"io.delta" % "delta-contribs_2.12" % "1.0.0",
"com.google.cloud.spark" % "spark-bigquery-with-dependencies_2.12" % "0.21.1",
"com.google.cloud.bigdataoss" % "gcs-connector" % "1.9.4-hadoop3"

在 GCS 中检查我的增量文件：

$ gsutil ls gs://r-dps-datapipeline-dev/testoliver/oliver_sample_delta3
gs://r-dps-datapipeline-dev/testoliver/oliver_sample_delta3/
gs://r-dps-datapipeline-dev/testoliver/oliver_sample_delta3/part-00000-ce79bfc7-e28f-4929-955c-56a7a08caf9f-c000.snappy.parquet
gs://r-dps-datapipeline-dev/testoliver/oliver_sample_delta3/part-00001-dda0bd2d-a081-4444-8983-ac8f3a2ffe9d-c000.snappy.parquet
gs://r-dps-datapipeline-dev/testoliver/oliver_sample_delta3/part-00002-93f7429b-777a-42f4-b2dd-adc9a482a6e8-c000.snappy.parquet
gs://r-dps-datapipeline-dev/testoliver/oliver_sample_delta3/part-00003-e9874baf-6c0b-46de-891e-032ac8b67287-c000.snappy.parquet
gs://r-dps-datapipeline-dev/testoliver/oliver_sample_delta3/part-00004-ede54816-2da1-412f-a9e3-5233e77258fb-c000.snappy.parquet
gs://r-dps-datapipeline-dev/testoliver/oliver_sample_delta3/_delta_log/
gs://r-dps-datapipeline-dev/testoliver/oliver_sample_delta3/_symlink_format_manifest/

【讨论】：