使用s3a协议从本地pyspark上的s3读取Parquet文件时发生异常

mw3dktmi  于 2021-05-29  发布在  Spark
关注(0)|答案(0)|浏览(342)

下面有一个简单的代码，可以使用s3a从s3读取Parquet文件

  1. import os
  2. from pyspark.sql import SparkSession
  3. import configparser
  5. # os.environ['PYSPARK_SUBMIT_ARGS'] = "--packages=org.apache.hadoop:hadoop-aws:2.7.3 pyspark-shell"
  7. # Get the AWS credentials for the mlandexps profile using configparser library
  9. config = configparser.ConfigParser()
  10. config.read(os.path.expanduser("~/.aws/credentials"))
  12. access_key = config.get("mlandexps", "aws_access_key_id")
  13. secret_key = config.get("mlandexps", "aws_secret_access_key")
  15. print("Access Key : %s" % access_key)
  16. print("Secret Key : %s" % secret_key)
  18. # Create a Spark session
  20. spark = SparkSession \
  21.     .builder \
  22.     .master("local") \
  23.     .appName("ReadFilesFromS3TestUsingS3a") \
  24.     .getOrCreate()
  26. sc = spark.sparkContext
  28. hadoop_conf=sc._jsc.hadoopConfiguration()
  29. hadoop_conf.set("fs.s3a.endpoint", "s3.amazonaws.com")
  30. hadoop_conf.set("fs.s3a.access.key", access_key)
  31. hadoop_conf.set("fs.s3a.secret.key", secret_key)
  32. hadoop_conf.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem");
  34. df = spark.read.parquet("s3a://rppatwa/BlackWidowData/part-00000-5b22d5ed-2be1-4343-9aa2-53e2133916ae-c000.snappy.parquet")
  35. df.printSchema()
  36. df.show(3)

当我在本地运行它时，会出现以下异常

  1. Py4JJavaError                             Traceback (most recent call last)
  3. <ipython-input-17-ffd16a1a558f> in <module>
  4.      37 print("Spark Context : {}".format(sc))
  5.      38 
  6. ---> 39 df = spark.read.parquet("s3a://rppatwa/BlackWidowData/part-00000-5b22d5ed-2be1-4343-9aa2-53e2133916ae-c000.snappy.parquet")
  7.      40 df.printSchema()
  8.      41 df.show(3)
  10. ~/Downloads/Spark/spark-2.4.5-bin-hadoop2.7/python/pyspark/sql/readwriter.py in parquet(self, *paths)
  11.     314         [('name', 'string'), ('year', 'int'), ('month', 'int'), ('day', 'int')]
  12.     315         """
  13. --> 316         return self._df(self._jreader.parquet(_to_seq(self._spark._sc, paths)))
  14.     317 
  15.     318     @ignore_unicode_prefix
  17. ~/Downloads/Spark/spark-2.4.5-bin-hadoop2.7/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py in __call__(self, *args)
  18.    1255         answer = self.gateway_client.send_command(command)
  19.    1256         return_value = get_return_value(
  20. -> 1257             answer, self.gateway_client, self.target_id, self.name)
  21.    1258 
  22.    1259         for temp_arg in temp_args:
  24. ~/Downloads/Spark/spark-2.4.5-bin-hadoop2.7/python/pyspark/sql/utils.py in deco(*a,**kw)
  25.      61     def deco(*a,**kw):
  26.      62         try:
  27. ---> 63             return f(*a,**kw)
  28.      64         except py4j.protocol.Py4JJavaError as e:
  29.      65             s = e.java_exception.toString()
  31. ~/Downloads/Spark/spark-2.4.5-bin-hadoop2.7/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
  32.     326                 raise Py4JJavaError(
  33.     327                     "An error occurred while calling {0}{1}{2}.\n".
  34. --> 328                     format(target_id, ".", name), value)
  35.     329             else:
  36.     330                 raise Py4JError(
  38. Py4JJavaError: An error occurred while calling o514.parquet.
  39. : org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 7.0 failed 1 times, most recent failure: Lost task 0.0 in stage 7.0 (TID 7, localhost, executor driver): java.io.IOException: Failed to connect to /192.168.19.237:62286
  40.     at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:245)
  41.     at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:187)
  42.     at org.apache.spark.rpc.netty.NettyRpcEnv.org$apache$spark$rpc$netty$NettyRpcEnv$$downloadClient(NettyRpcEnv.scala:368)
  43.     at org.apache.spark.rpc.netty.NettyRpcEnv$$anonfun$openChannel$1.apply$mcV$sp(NettyRpcEnv.scala:336)
  44.     at org.apache.spark.rpc.netty.NettyRpcEnv$$anonfun$openChannel$1.apply(NettyRpcEnv.scala:335)
  45.     at org.apache.spark.rpc.netty.NettyRpcEnv$$anonfun$openChannel$1.apply(NettyRpcEnv.scala:335)
  46.     at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1394)
  47.     at org.apache.spark.rpc.netty.NettyRpcEnv.openChannel(NettyRpcEnv.scala:339)
  48.     at org.apache.spark.util.Utils$.doFetchFile(Utils.scala:693)
  49.     at org.apache.spark.util.Utils$.fetchFile(Utils.scala:509)
  50.     at org.apache.spark.executor.Executor$$anonfun$org$apache$spark$executor$Executor$$updateDependencies$5.apply(Executor.scala:816)
  51.     at org.apache.spark.executor.Executor$$anonfun$org$apache$spark$executor$Executor$$updateDependencies$5.apply(Executor.scala:808)
  52.     at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733)
  53.     at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:130)
  54.     at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:130)
  55.     at scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:236)
  56.     at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:40)
  57.     at scala.collection.mutable.HashMap.foreach(HashMap.scala:130)
  58.     at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732)
  59.     at org.apache.spark.executor.Executor.org$apache$spark$executor$Executor$$updateDependencies(Executor.scala:808)
  60.     at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:375)
  61.     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  62.     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  63.     at java.lang.Thread.run(Thread.java:748)
  64.     Suppressed: java.lang.NullPointerException
  65.         at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1402)
  66.         ... 17 more
  67. Caused by: io.netty.channel.AbstractChannel$AnnotatedConnectException: Operation timed out: /192.168.19.237:62286
  68. Caused by: java.net.ConnectException: Operation timed out
  69.     at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
  70.     at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)
  71.     at io.netty.channel.socket.nio.NioSocketChannel.doFinishConnect(NioSocketChannel.java:327)
  72.     at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:334)
  73.     at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:688)
  74.     at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:635)
  75.     at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:552)
  76.     at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:514)
  77.     at io.netty.util.concurrent.SingleThreadEventExecutor$6.run(SingleThreadEventExecutor.java:1044)
  78.     at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
  79.     at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
  80.     at java.lang.Thread.run(Thread.java:748)
  82. Driver stacktrace:
  83.     at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1891)
  84.     at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1879)
  85.     at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1878)
  86.     at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
  87.     at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
  88.     at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1878)
  89.     at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:927)
  90.     at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:927)
  91.     at scala.Option.foreach(Option.scala:257)
  92.     at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:927)
  93.     at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2112)
  94.     at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2061)
  95.     at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2050)
  96.     at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
  97.     at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:738)
  98.     at org.apache.spark.SparkContext.runJob(SparkContext.scala:2061)
  99.     at org.apache.spark.SparkContext.runJob(SparkContext.scala:2082)
  100.     at org.apache.spark.SparkContext.runJob(SparkContext.scala:2101)
  101.     at org.apache.spark.SparkContext.runJob(SparkContext.scala:2126)
  102.     at org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:990)
  103.     at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  104.     at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
  105.     at org.apache.spark.rdd.RDD.withScope(RDD.scala:385)
  106.     at org.apache.spark.rdd.RDD.collect(RDD.scala:989)
  107.     at org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat$.mergeSchemasInParallel(ParquetFileFormat.scala:633)
  108.     at org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat.inferSchema(ParquetFileFormat.scala:241)
  109.     at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$6.apply(DataSource.scala:180)
  110.     at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$6.apply(DataSource.scala:180)
  111.     at scala.Option.orElse(Option.scala:289)
  112.     at org.apache.spark.sql.execution.datasources.DataSource.getOrInferFileFormatSchema(DataSource.scala:179)
  113.     at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:373)
  114.     at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:223)
  115.     at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:211)
  116.     at org.apache.spark.sql.DataFrameReader.parquet(DataFrameReader.scala:645)
  117.     at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  118.     at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  119.     at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  120.     at java.lang.reflect.Method.invoke(Method.java:498)
  121.     at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
  122.     at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
  123.     at py4j.Gateway.invoke(Gateway.java:282)
  124.     at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
  125.     at py4j.commands.CallCommand.execute(CallCommand.java:79)
  126.     at py4j.GatewayConnection.run(GatewayConnection.java:238)
  127.     at java.lang.Thread.run(Thread.java:748)
  128. Caused by: java.io.IOException: Failed to connect to /192.168.19.237:62286
  129.     at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:245)
  130.     at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:187)
  131.     at org.apache.spark.rpc.netty.NettyRpcEnv.org$apache$spark$rpc$netty$NettyRpcEnv$$downloadClient(NettyRpcEnv.scala:368)
  132.     at org.apache.spark.rpc.netty.NettyRpcEnv$$anonfun$openChannel$1.apply$mcV$sp(NettyRpcEnv.scala:336)
  133.     at org.apache.spark.rpc.netty.NettyRpcEnv$$anonfun$openChannel$1.apply(NettyRpcEnv.scala:335)
  134.     at org.apache.spark.rpc.netty.NettyRpcEnv$$anonfun$openChannel$1.apply(NettyRpcEnv.scala:335)
  135.     at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1394)
  136.     at org.apache.spark.rpc.netty.NettyRpcEnv.openChannel(NettyRpcEnv.scala:339)
  137.     at org.apache.spark.util.Utils$.doFetchFile(Utils.scala:693)
  138.     at org.apache.spark.util.Utils$.fetchFile(Utils.scala:509)
  139.     at org.apache.spark.executor.Executor$$anonfun$org$apache$spark$executor$Executor$$updateDependencies$5.apply(Executor.scala:816)
  140.     at org.apache.spark.executor.Executor$$anonfun$org$apache$spark$executor$Executor$$updateDependencies$5.apply(Executor.scala:808)
  141.     at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733)
  142.     at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:130)
  143.     at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:130)
  144.     at scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:236)
  145.     at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:40)
  146.     at scala.collection.mutable.HashMap.foreach(HashMap.scala:130)
  147.     at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732)
  148.     at org.apache.spark.executor.Executor.org$apache$spark$executor$Executor$$updateDependencies(Executor.scala:808)
  149.     at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:375)
  150.     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  151.     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  152.     ... 1 more
  153.     Suppressed: java.lang.NullPointerException
  154.         at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1402)
  155.         ... 17 more
  156. Caused by: io.netty.channel.AbstractChannel$AnnotatedConnectException: Operation timed out: /192.168.19.237:62286
  157. Caused by: java.net.ConnectException: Operation timed out
  158.     at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
  159.     at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)
  160.     at io.netty.channel.socket.nio.NioSocketChannel.doFinishConnect(NioSocketChannel.java:327)
  161.     at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:334)
  162.     at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:688)
  163.     at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:635)
  164.     at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:552)
  165.     at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:514)
  166.     at io.netty.util.concurrent.SingleThreadEventExecutor$6.run(SingleThreadEventExecutor.java:1044)
  167.     at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
  168.     at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
  169.     at java.lang.Thread.run(Thread.java:748)

我猜这个异常的根本原因和这行有关
原因：java.io.ioexception:无法连接到/192.168.19.237:62286
我环顾了一下网络，没有找到一个非常有用的答案。感谢你的观点！
谢谢

hadooppythonapache-sparkpysparkamazon-s3

暂无答案！

目前还没有任何答案，快来回答吧！

我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com