结合delta io和excel阅读

Question

spark = SparkSession.builder.appName("Word Count")
.config("spark.jars.packages", "com.crealytics:spark-excel_2.12:0.14.0")
.getOrCreate()

df = spark.read.format("com.crealytics.spark.excel")
.option("header", "true")
.load(path2)

builder = SparkSession.builder.appName("transaction")
.config("spark.jars.packages", "com.crealytics:spark-excel_2.12:0.14.0")
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")

spark = configure_spark_with_delta_pip(builder).getOrCreate()

给出以下错误：

PY4JJavaError：调用O139.Load时出错。：java.lang.ClassNotFoundException：未能找到html" target="_blank">数据源：com.crealytics.spark.excel。请在http://spark.apache.org/third-party-projects.html找到包，地址为org.apache.spark.sql.execution.datasources.datasource$.lookupdatasource(datasource.scala:692)地址为org.apache.spark.sql.execution.datasources.datasource$.lookupdatasource2(datasource.scala:746)地址为org.apache.spark.sql.dataframereader.load（239）在java.base/jdk.internal.reflect.nativeMethodAccessorImpl.Invoke0（原生方法）在java.base/jdk.internal.reflect.nativeMethodAccessorImpl.invoke(nativeMethodAccessorImpl.invoke(nativeMethodAccessorImpl.java:62)在在py4j.gateway.invoke(gateway.java:282)在py4j.commands.abstractCommand.invokeMethod（AbstractCommand.java:132)在py4j.commands.callcommand.execute(callcommand.java:79)在py4j.gatewayConnection.run（gatewayConnection.java:238)在java.base/java.lang.thread.run（thread.java:829)在java.base/java.lang.thread.run（thread.java:829)在java.base/java.net.urlClassLoader.findclass（urlClassLoader.java:471)在spark.sql.execution.datasources.datasource$.$$anonfun$lookupdatasource$5(datasource.scala:666)at scala.util.try$.applic(try.scala:213)at org.apache.spark.sql.execution.datasources.datasource$.$$anonfun$lookupdatasource$4(datasource.scala:666)at scal.apache.spark.sql.execution.datasource:666)at

为什么？我该怎么避免呢？

Answer 1

之所以出现此错误，是因为configure_spark_with_delta_pip覆盖/替换了要导入适当的delta lake包的配置属性spark.jars.packages。因此，您的包com.crealytics:spark-excel2.12:0.14.0可能不可用/无法导入。在这里查看源代码的一个片段

    scala_version = "2.12"
    maven_artifact = f"io.delta:delta-core_{scala_version}:{delta_version}"

    return spark_session_builder.config("spark.jars.packages", maven_artifact) 

不幸的是，构建器不允许我们检索现有的配置属性或sparkconf对象，以便在调用getorcreate创建或触发会话之前动态调整这些属性。

要解决这个问题，您可以自己检索适当的delta包，类似于configure_spark_with_delta_pip这样做的方式。

import importlib_metadata
delta_version = importlib_metadata.version("delta_spark")
scala_version = "2.12"
delta_package = f"io.delta:delta-core_{scala_version}:{delta_version}"

builder = SparkSession.builder.appName("transaction")
.config("spark.jars.packages", f"com.crealytics:spark-excel_2.12:0.14.0,{delta_package}")
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")

spark = configure_spark_with_delta_pip(builder).getOrCreate()

要解决这个问题，可以在使用configure_spark_with_delta_pip应用delta包后创建spark会话。之后，您可以使用更新的配置属性触发spark会话的重新启动。

如。

builder = SparkSession.builder.appName("transaction")
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")

spark = configure_spark_with_delta_pip(builder).getOrCreate()

builder = SparkSession.builder.appName("transaction")
.config("spark.jars.packages", "com.crealytics:spark-excel_2.12:0.14.0")

spark = builder.getOrCreate()

由于两个spark会话都具有相同的appname，getorcreate将检索现有的spark会话，但也将应用新的配置。此行为在此记录为

>>> s2 = SparkSession.builder.config("k2", "v2").getOrCreate()
>>> s1.conf.get("k1") == s2.conf.get("k1") 
True
>>> s1.conf.get("k2") == s2.conf.get("k2") 
True