如何解决通过 pyspark 流式传输数据时出现 UnsupportedOperationException
我正在使用这段简单的代码从目录中读取 json 文件流。该代码在 Databricks 笔记本上运行良好,但在本地运行时抛出错误。我正在使用 databricks-connect(8.1 版)通过集群连接和运行脚本。
from pyspark.sql.types import StructType
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("ProcessSensorData").getorCreate()
userschema = StructType().add("ID","string").add("Created","string")\
.add("Data","string").add("deviceid","string").add("Size","string")
df = spark.readStream.schema(userschema).json("dbfs:/mnt/")
df.writeStream.format("parquet")\
.option("checkpointLocation","dbfs:/mnt/parquet/demo_checkpoint1")\
.option("path","dbfs:/mnt/parquet/demo_parquet1")\
.start()
当我使用“read”而不是“readStream”时,上面的代码在本地运行良好。我尝试使用不同的方式使用选项、格式来读取流,并且还确认了我与数据块集群的连接。我有 pyspark 版本 3.1.1 和 java 8。我总是收到以下错误:
21/04/21 09:10:44 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR,use setLogLevel(newLevel).
21/04/21 09:10:45 WARN MetricsSystem: Using default name SparkStatusTracker for source because neither spark.metrics.namespace nor spark.app.id is set.
Traceback (most recent call last):
File "/Users/dir/spark_process.py",line 6,in <module>
df = spark.readStream.schema(userschema).json("dbfs:/mnt/")
File "/Users/dir/venv/lib/python3.9/site-packages/pyspark/sql/streaming.py",line 631,in json
return self._df(self._jreader.json(path))
File "/Users/dir/venv/lib/python3.9/site-packages/py4j/java_gateway.py",line 1304,in __call__
return_value = get_return_value(
File "/Users/dir/venv/lib/python3.9/site-packages/pyspark/sql/utils.py",line 110,in deco
return f(*a,**kw)
File "/Users/dir/venv/lib/python3.9/site-packages/py4j/protocol.py",line 326,in get_return_value
raise Py4JJavaError(
py4j.protocol.Py4JJavaError: An error occurred while calling o31.json.
: java.lang.UnsupportedOperationException
at com.databricks.sql.transaction.directory.DirectoryAtomicReadProtocol$.filterDirectoryListing(DirectoryAtomicReadProtocol.scala:28)
at org.apache.spark.sql.execution.datasources.InMemoryFileIndex$.listLeafFiles(InMemoryFileIndex.scala:375)
at org.apache.spark.sql.execution.datasources.InMemoryFileIndex$.$anonfun$bulkListLeafFiles$2(InMemoryFileIndex.scala:282)
at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:238)
at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
at scala.collection.TraversableLike.map(TraversableLike.scala:238)
at scala.collection.TraversableLike.map$(TraversableLike.scala:231)
at scala.collection.AbstractTraversable.map(Traversable.scala:108)
at org.apache.spark.sql.execution.datasources.InMemoryFileIndex$.bulkListLeafFiles(InMemoryFileIndex.scala:274)
at org.apache.spark.sql.execution.datasources.InMemoryFileIndex.listLeafFiles(InMemoryFileIndex.scala:139)
at org.apache.spark.sql.execution.datasources.InMemoryFileIndex.refresh0(InMemoryFileIndex.scala:102)
at org.apache.spark.sql.execution.datasources.InMemoryFileIndex.<init>(InMemoryFileIndex.scala:74)
at org.apache.spark.sql.execution.datasources.DataSource.createInMemoryFileIndex(DataSource.scala:620)
at org.apache.spark.sql.execution.datasources.DataSource.$anonfun$sourceSchema$2(DataSource.scala:296)
at org.apache.spark.sql.execution.datasources.DataSource.tempFileIndex$lzycompute$1(DataSource.scala:183)
at org.apache.spark.sql.execution.datasources.DataSource.tempFileIndex$1(DataSource.scala:183)
at org.apache.spark.sql.execution.datasources.DataSource.getorInferFileFormatSchema(DataSource.scala:188)
at org.apache.spark.sql.execution.datasources.DataSource.sourceSchema(DataSource.scala:288)
at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo$lzycompute(DataSource.scala:137)
at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo(DataSource.scala:137)
at org.apache.spark.sql.execution.streaming.Streamingrelation$.apply(Streamingrelation.scala:33)
at org.apache.spark.sql.streaming.DataStreamReader.loadInternal(DataStreamReader.scala:264)
at org.apache.spark.sql.streaming.DataStreamReader.load(DataStreamReader.scala:280)
at org.apache.spark.sql.streaming.DataStreamReader.json(DataStreamReader.scala:361)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:380)
at py4j.Gateway.invoke(Gateway.java:295)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:251)
at java.lang.Thread.run(Thread.java:748)
Process finished with exit code 1
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。