python - pyspark:返回不完整的 URI 错误

Question

我正在尝试使用 HDFS 中的文件加载 MongoDB 表。我一直在寻找类似的问题，我发现 HDFS URI 必须有三个栏。但是，即使我将 HDFS URI 设置为三个条，我也会遇到同样的错误:

这是我的代码:

from pyspark.sql import functions
from pyspark.sql import Row
from pyspark.sql import SparkSession

def parseInput(line):
    fields=line.split('|')
    return Row(userId=int(fields[0]), age=int(fields[1]), gender=fields[2], occupation=fields[3], zip=fields)

if __name__ == '__main__':
    spark=SparkSession.builder.appName('Monark').getOrCreate()
    lines=spark.sparkContext.textFile('hdfs:///user/hadoop/u.user')

    users=lines.map(parseInput)


    usersDataset=spark.createDataFrame(users)

    usersDataset.write\
    .format("com.mongodb.spark.sql.DefaultSource")\
    .mode("append")\
    .options("uri","mongodb://127.0.0.1/movielens.users")\
    .save()

    readUsers.read=spark.read\
    .format("uri","mongodb://127.0.0.1/movielens.users")\
    .mode("append")\
    .options()\
    .load()

    readUsers.readOrReplaceTempView("users")

    sqlDf=spark.sql("SELECT * FROM users where age < 20")

    sqlDf.show()

    spark.stop()

执行日志:

Ivy Default Cache set to: /home/hadoop/.ivy2/cache
The jars for the packages stored in: /home/hadoop/.ivy2/jars
:: loading settings :: url = jar:file:/usr/local/lib/python2.7/dist-packages/pyspark/jars/ivy-2.4.0.jar!/org/apache/ivy/core/settings/ivysettings.xml
org.mongodb.spark#mongo-spark-connector_2.11 added as a dependency
:: resolving dependencies :: org.apache.spark#spark-submit-parent;1.0
    confs: [default]
    found org.mongodb.spark#mongo-spark-connector_2.11;2.0.0 in central
    found org.mongodb#mongo-java-driver;3.2.2 in central
:: resolution report :: resolve 653ms :: artifacts dl 21ms
    :: modules in use:
    org.mongodb#mongo-java-driver;3.2.2 from central in [default]
    org.mongodb.spark#mongo-spark-connector_2.11;2.0.0 from central in [default]
    ---------------------------------------------------------------------
    |                  |            modules            ||   artifacts   |
    |       conf       | number| search|dwnlded|evicted|| number|dwnlded|
    ---------------------------------------------------------------------
    |      default     |   2   |   0   |   0   |   0   ||   2   |   0   |
    ---------------------------------------------------------------------
:: retrieving :: org.apache.spark#spark-submit-parent
    confs: [default]
    0 artifacts copied, 2 already retrieved (0kB/42ms)
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
18/03/24 03:45:48 WARN Utils: Your hostname, ubuntu resolves to a loopback address: 127.0.0.1; using 192.168.56.145 instead (on interface ens33)
18/03/24 03:45:48 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address
18/03/24 03:45:49 INFO SparkContext: Running Spark version 2.2.1
18/03/24 03:45:50 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
18/03/24 03:45:50 INFO SparkContext: Submitted application: Monark
18/03/24 03:45:50 INFO SecurityManager: Changing view acls to: hadoop
18/03/24 03:45:50 INFO SecurityManager: Changing modify acls to: hadoop
18/03/24 03:45:50 INFO SecurityManager: Changing view acls groups to: 
18/03/24 03:45:50 INFO SecurityManager: Changing modify acls groups to: 
18/03/24 03:45:50 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users  with view permissions: Set(hadoop); groups with view permissions: Set(); users  with modify permissions: Set(hadoop); groups with modify permissions: Set()
18/03/24 03:45:51 INFO Utils: Successfully started service 'sparkDriver' on port 35259.
18/03/24 03:45:52 INFO SparkEnv: Registering MapOutputTracker
18/03/24 03:45:52 INFO SparkEnv: Registering BlockManagerMaster
18/03/24 03:45:52 INFO BlockManagerMasterEndpoint: Using org.apache.spark.storage.DefaultTopologyMapper for getting topology information
18/03/24 03:45:52 INFO BlockManagerMasterEndpoint: BlockManagerMasterEndpoint up
18/03/24 03:45:52 INFO DiskBlockManager: Created local directory at /tmp/blockmgr-082ae340-d5bd-4e90-973c-425f52e0ba71
18/03/24 03:45:52 INFO MemoryStore: MemoryStore started with capacity 413.9 MB
18/03/24 03:45:52 INFO SparkEnv: Registering OutputCommitCoordinator
18/03/24 03:45:53 INFO Utils: Successfully started service 'SparkUI' on port 4040.
18/03/24 03:45:53 INFO SparkUI: Bound SparkUI to 0.0.0.0, and started at http://192.168.56.145:4040
18/03/24 03:45:53 INFO SparkContext: Added JAR file:/home/hadoop/.ivy2/jars/org.mongodb.spark_mongo-spark-connector_2.11-2.0.0.jar at spark://192.168.56.145:35259/jars/org.mongodb.spark_mongo-spark-connector_2.11-2.0.0.jar with timestamp 1521888353694
18/03/24 03:45:53 INFO SparkContext: Added JAR file:/home/hadoop/.ivy2/jars/org.mongodb_mongo-java-driver-3.2.2.jar at spark://192.168.56.145:35259/jars/org.mongodb_mongo-java-driver-3.2.2.jar with timestamp 1521888353701
18/03/24 03:45:54 INFO SparkContext: Added file file:/home/hadoop/Monark.py at file:/home/hadoop/Monark.py with timestamp 1521888354652
18/03/24 03:45:54 INFO Utils: Copying /home/hadoop/Monark.py to /tmp/spark-5826c09b-085e-467a-8a60-43814d565820/userFiles-01e51ac3-60ec-4044-935b-b4df7f9e5732/Monark.py
18/03/24 03:45:54 INFO SparkContext: Added file file:/home/hadoop/.ivy2/jars/org.mongodb.spark_mongo-spark-connector_2.11-2.0.0.jar at file:/home/hadoop/.ivy2/jars/org.mongodb.spark_mongo-spark-connector_2.11-2.0.0.jar with timestamp 1521888354792
18/03/24 03:45:54 INFO Utils: Copying /home/hadoop/.ivy2/jars/org.mongodb.spark_mongo-spark-connector_2.11-2.0.0.jar to /tmp/spark-5826c09b-085e-467a-8a60-43814d565820/userFiles-01e51ac3-60ec-4044-935b-b4df7f9e5732/org.mongodb.spark_mongo-spark-connector_2.11-2.0.0.jar
18/03/24 03:45:54 INFO SparkContext: Added file file:/home/hadoop/.ivy2/jars/org.mongodb_mongo-java-driver-3.2.2.jar at file:/home/hadoop/.ivy2/jars/org.mongodb_mongo-java-driver-3.2.2.jar with timestamp 1521888354848
18/03/24 03:45:54 INFO Utils: Copying /home/hadoop/.ivy2/jars/org.mongodb_mongo-java-driver-3.2.2.jar to /tmp/spark-5826c09b-085e-467a-8a60-43814d565820/userFiles-01e51ac3-60ec-4044-935b-b4df7f9e5732/org.mongodb_mongo-java-driver-3.2.2.jar
18/03/24 03:45:55 INFO Executor: Starting executor ID driver on host localhost
18/03/24 03:45:55 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 39795.
18/03/24 03:45:55 INFO NettyBlockTransferService: Server created on 192.168.56.145:39795
18/03/24 03:45:55 INFO BlockManager: Using org.apache.spark.storage.RandomBlockReplicationPolicy for block replication policy
18/03/24 03:45:55 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(driver, 192.168.56.145, 39795, None)
18/03/24 03:45:55 INFO BlockManagerMasterEndpoint: Registering block manager 192.168.56.145:39795 with 413.9 MB RAM, BlockManagerId(driver, 192.168.56.145, 39795, None)
18/03/24 03:45:55 INFO BlockManagerMaster: Registered BlockManager BlockManagerId(driver, 192.168.56.145, 39795, None)
18/03/24 03:45:55 INFO BlockManager: Initialized BlockManager: BlockManagerId(driver, 192.168.56.145, 39795, None)
18/03/24 03:45:56 INFO SharedState: Setting hive.metastore.warehouse.dir ('null') to the value of spark.sql.warehouse.dir ('file:/home/hadoop/spark-warehouse/').
18/03/24 03:45:56 INFO SharedState: Warehouse path is 'file:/home/hadoop/spark-warehouse/'.
18/03/24 03:45:57 INFO StateStoreCoordinatorRef: Registered StateStoreCoordinator endpoint
18/03/24 03:45:59 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 238.6 KB, free 413.7 MB)
18/03/24 03:45:59 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 22.9 KB, free 413.7 MB)
18/03/24 03:45:59 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on 192.168.56.145:39795 (size: 22.9 KB, free: 413.9 MB)
18/03/24 03:45:59 INFO SparkContext: Created broadcast 0 from textFile at NativeMethodAccessorImpl.java:0
Traceback (most recent call last):
  File "/home/hadoop/Monark.py", line 16, in <module>
    usersDataset=spark.createDataFrame(users)
  File "/usr/local/lib/python2.7/dist-packages/pyspark/python/lib/pyspark.zip/pyspark/sql/session.py", line 582, in createDataFrame
  File "/usr/local/lib/python2.7/dist-packages/pyspark/python/lib/pyspark.zip/pyspark/sql/session.py", line 380, in _createFromRDD
  File "/usr/local/lib/python2.7/dist-packages/pyspark/python/lib/pyspark.zip/pyspark/sql/session.py", line 351, in _inferSchema
  File "/usr/local/lib/python2.7/dist-packages/pyspark/python/lib/pyspark.zip/pyspark/rdd.py", line 1361, in first
  File "/usr/local/lib/python2.7/dist-packages/pyspark/python/lib/pyspark.zip/pyspark/rdd.py", line 1313, in take
  File "/usr/local/lib/python2.7/dist-packages/pyspark/python/lib/pyspark.zip/pyspark/rdd.py", line 2440, in getNumPartitions
  File "/usr/local/lib/python2.7/dist-packages/pyspark/python/lib/py4j-0.10.4-src.zip/py4j/java_gateway.py", line 1133, in __call__
  File "/usr/local/lib/python2.7/dist-packages/pyspark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 63, in deco
  File "/usr/local/lib/python2.7/dist-packages/pyspark/python/lib/py4j-0.10.4-src.zip/py4j/protocol.py", line 319, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o30.partitions.
: java.io.IOException: Incomplete HDFS URI, no host: hdfs:/user/hadoop/u.user
    at org.apache.hadoop.hdfs.DistributedFileSystem.initialize(DistributedFileSystem.java:143)
    at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2669)
    at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:94)
    at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:2703)
    at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:2685)
    at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:373)
    at org.apache.hadoop.fs.Path.getFileSystem(Path.java:295)
    at org.apache.hadoop.mapred.FileInputFormat.singleThreadedListStatus(FileInputFormat.java:258)
    at org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:229)
    at org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:315)
    at org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:199)
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:252)
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:250)
    at scala.Option.getOrElse(Option.scala:121)
    at org.apache.spark.rdd.RDD.partitions(RDD.scala:250)
    at org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35)
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:252)
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:250)
    at scala.Option.getOrElse(Option.scala:121)
    at org.apache.spark.rdd.RDD.partitions(RDD.scala:250)
    at org.apache.spark.api.java.JavaRDDLike$class.partitions(JavaRDDLike.scala:61)
    at org.apache.spark.api.java.AbstractJavaRDDLike.partitions(JavaRDDLike.scala:45)
    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:357)
    at py4j.Gateway.invoke(Gateway.java:280)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:214)
    at java.lang.Thread.run(Thread.java:748)

18/03/24 03:46:00 INFO SparkContext: Invoking stop() from shutdown hook
18/03/24 03:46:00 INFO SparkUI: Stopped Spark web UI at http://192.168.56.145:4040
18/03/24 03:46:00 INFO MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped!
18/03/24 03:46:00 INFO MemoryStore: MemoryStore cleared
18/03/24 03:46:00 INFO BlockManager: BlockManager stopped
18/03/24 03:46:00 INFO BlockManagerMaster: BlockManagerMaster stopped
18/03/24 03:46:00 INFO OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped!
18/03/24 03:46:00 INFO SparkContext: Successfully stopped SparkContext
18/03/24 03:46:00 INFO ShutdownHookManager: Shutdown hook called
18/03/24 03:46:00 INFO ShutdownHookManager: Deleting directory /tmp/spark-5826c09b-085e-467a-8a60-43814d565820
18/03/24 03:46:00 INFO ShutdownHookManager: Deleting directory /tmp/spark-5826c09b-085e-467a-8a60-43814d565820/pyspark-0bece44d-4127-47bd-8391-98d4da0cf695

查看执行日志，可以看到另外两个bar都被去掉了，不知道为什么会这样:

py4j.protocol.Py4JJavaError:调用 o30.partitions 时出错。: java.io.IOException: 不完整的 HDFS URI，没有主机:hdfs:/user/hadoop/u.user

谢谢大家

Answer 1

打开core-site.xml文件并在其中输入以下内容。

<property>
   <name>fs.default.name</name>
   <value>hdfs://localhost:54310</value>
   <description>The name of the default file system.  A URI whose scheme and authority determine the FileSystem implementation.  The uri's scheme determines the config $
 </property>

并更新 lines=spark.sparkContext.textFile('hdfs://localhost:54310/user/hadoop/u.user') 而不是 lines=spark.sparkContext.textFile ('hdfs:///user/hadoop/u.user')