DF.topandas() - 无法在 hadoop 二进制路径中找到 winutils 二进制文件

Posted

技术标签:

【中文标题】DF.topandas() - 无法在 hadoop 二进制路径中找到 winutils 二进制文件【英文标题】:DF.topandas() - Failed to locate the winutils binary in the hadoop binary path 【发布时间】:2018-11-18 03:40:45 【问题描述】:

我正在使用 PyCharm 和 PySpark 运行一个巨大的文本文件。

这就是我想要做的:

spark_home = os.environ.get('SPARK_HOME', None)
os.environ["SPARK_HOME"] = "C:\spark-2.3.0-bin-hadoop2.7"
import pyspark
from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession
conf = SparkConf()
sc = SparkContext(conf=conf)
spark = SparkSession.builder.config(conf=conf).getOrCreate() 
import pandas as pd
ip = spark.read.format("csv").option("inferSchema","true").option("header","true").load(r"some other file.csv")
kw = pd.read_csv(r"some file.csv",encoding='ISO-8859-1',index_col=False,error_bad_lines=False)
for i in range(len(kw)):
    rx = '(?i)'+kw.Keywords[i]
    ip = ip.where(~ip['Content'].rlike(rx))
op = ip.toPandas()
op.to_csv(r'something.csv',encoding='utf-8')

但是,PyCharm 向我抛出了这个错误:

To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
2018-06-08 11:31:52 WARN  Utils:66 - Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.debug.maxToStringFields' in SparkEnv.conf.
Traceback (most recent call last):
  File "C:/Users/mainak.paul/PycharmProjects/Concept_Building_SIP/ThemeSparkUncoveredGames.py", line 17, in <module>
    op = ip.toPandas()
  File "C:\Python27\lib\site-packages\pyspark\sql\dataframe.py", line 1966, in toPandas
    pdf = pd.DataFrame.from_records(self.collect(), columns=self.columns)
  File "C:\Python27\lib\site-packages\pyspark\sql\dataframe.py", line 466, in collect
    port = self._jdf.collectToPython()
  File "C:\Python27\lib\site-packages\py4j\java_gateway.py", line 1160, in __call__
    answer, self.gateway_client, self.target_id, self.name)
  File "C:\Python27\lib\site-packages\pyspark\sql\utils.py", line 63, in deco
    return f(*a, **kw)
  File "C:\Python27\lib\site-packages\py4j\protocol.py", line 320, in get_return_value
    format(target_id, ".", name), value)
py4j.protocol.Py4JJavaError: An error occurred while calling o30.collectToPython.
: java.lang.IllegalArgumentException

我只是不明白为什么.toPandas() 不起作用。 Spark 版本是 2.3。这个版本有什么我不知道的变化吗?我用 spark 2.2 在另一台机器上运行了这段代码,运行良好。

我什至把导出行改成了这样的

op = ip.where(ip['Content'].rlike(rx)).toPandas()

仍然遇到同样的错误。我究竟做错了什么?有没有其他方法可以在不影响性能的情况下将pyspark.sql.dataframe.DataFrame 导出到.csv

已编辑 我也试过使用:

ip.write.csv('file.csv')

现在我收到以下错误:

Traceback (most recent call last):
  File "somefile.csv", line 21, in <module>
    ip.write.csv('somefile.csv')
  File "C:\Python27\lib\site-packages\pyspark\sql\readwriter.py", line 883, in csv
    self._jwrite.csv(path)
  File "C:\Python27\lib\site-packages\py4j\java_gateway.py", line 1160, in __call__
    answer, self.gateway_client, self.target_id, self.name)
  File "C:\Python27\lib\site-packages\pyspark\sql\utils.py", line 63, in deco
    return f(*a, **kw)
  File "C:\Python27\lib\site-packages\py4j\protocol.py", line 320, in get_return_value
    format(target_id, ".", name), value)
py4j.protocol.Py4JJavaError: An error occurred while calling o102.csv.

添加堆栈跟踪:

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).
18/06/11 16:53:14 ERROR Shell: Failed to locate the winutils binary in the hadoop binary path
java.io.IOException: Could not locate executable C:\spark-2.3.0-bin-hadoop2.7\bin\bin\winutils.exe in the Hadoop binaries.
    at org.apache.hadoop.util.Shell.getQualifiedBinPath(Shell.java:379)
    at org.apache.hadoop.util.Shell.getWinUtilsPath(Shell.java:394)
    at org.apache.hadoop.util.Shell.<clinit>(Shell.java:387)
    at org.apache.hadoop.util.StringUtils.<clinit>(StringUtils.java:80)
    at org.apache.hadoop.security.SecurityUtil.getAuthenticationMethod(SecurityUtil.java:611)
    at org.apache.hadoop.security.UserGroupInformation.initialize(UserGroupInformation.java:273)
    at org.apache.hadoop.security.UserGroupInformation.ensureInitialized(UserGroupInformation.java:261)
    at org.apache.hadoop.security.UserGroupInformation.loginUserFromSubject(UserGroupInformation.java:791)
    at org.apache.hadoop.security.UserGroupInformation.getLoginUser(UserGroupInformation.java:761)
    at org.apache.hadoop.security.UserGroupInformation.getCurrentUser(UserGroupInformation.java:634)
    at org.apache.spark.util.Utils$$anonfun$getCurrentUserName$1.apply(Utils.scala:2430)
    at org.apache.spark.util.Utils$$anonfun$getCurrentUserName$1.apply(Utils.scala:2430)
    at scala.Option.getOrElse(Option.scala:121)
    at org.apache.spark.util.Utils$.getCurrentUserName(Utils.scala:2430)
    at org.apache.spark.SparkContext.<init>(SparkContext.scala:295)
    at org.apache.spark.api.java.JavaSparkContext.<init>(JavaSparkContext.scala:58)
    at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
    at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
    at java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:488)
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:247)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
    at py4j.Gateway.invoke(Gateway.java:236)
    at py4j.commands.ConstructorCommand.invokeConstructor(ConstructorCommand.java:80)
    at py4j.commands.ConstructorCommand.execute(ConstructorCommand.java:69)
    at py4j.GatewayConnection.run(GatewayConnection.java:214)
    at java.base/java.lang.Thread.run(Thread.java:844)
WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.apache.hadoop.security.authentication.util.KerberosUtil (file:/C:/opt/spark/spark-2.2.0-bin-hadoop2.7/jars/hadoop-auth-2.7.3.jar) to method sun.security.krb5.Config.getInstance()
WARNING: Please consider reporting this to the maintainers of org.apache.hadoop.security.authentication.util.KerberosUtil
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release
18/06/11 16:53:14 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Traceback (most recent call last):
  File "C:/Users/mainak.paul/PycharmProjects/Concept_Building_SIP/ThemeSparkUncoveredGames.py", line 22, in <module>
    op = ip.toPandas().collect()
  File "C:\Python27\lib\site-packages\pyspark\sql\dataframe.py", line 1937, in toPandas
    if self.sql_ctx.getConf("spark.sql.execution.pandas.respectSessionTimeZone").lower() \
  File "C:\Python27\lib\site-packages\pyspark\sql\context.py", line 142, in getConf
    return self.sparkSession.conf.get(key, defaultValue)
  File "C:\Python27\lib\site-packages\pyspark\sql\conf.py", line 46, in get
    return self._jconf.get(key)
  File "C:\Python27\lib\site-packages\py4j\java_gateway.py", line 1160, in __call__
    answer, self.gateway_client, self.target_id, self.name)
  File "C:\Python27\lib\site-packages\pyspark\sql\utils.py", line 63, in deco
    return f(*a, **kw)
  File "C:\Python27\lib\site-packages\py4j\protocol.py", line 320, in get_return_value
    format(target_id, ".", name), value)
py4j.protocol.Py4JJavaError: An error occurred while calling o86.get.
: java.util.NoSuchElementException: spark.sql.execution.pandas.respectSessionTimeZone
    at org.apache.spark.sql.internal.SQLConf$$anonfun$getConfString$2.apply(SQLConf.scala:1089)
    at org.apache.spark.sql.internal.SQLConf$$anonfun$getConfString$2.apply(SQLConf.scala:1089)
    at scala.Option.getOrElse(Option.scala:121)
    at org.apache.spark.sql.internal.SQLConf.getConfString(SQLConf.scala:1089)
    at org.apache.spark.sql.RuntimeConfig.get(RuntimeConfig.scala:74)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:564)
    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.base/java.lang.Thread.run(Thread.java:844)


Process finished with exit code 1

【问题讨论】:

【参考方案1】:

您需要按如下方式更改您的代码:

spark_home = os.environ.get('SPARK_HOME', None)
os.environ["SPARK_HOME"] = "C:\spark-2.3.0-bin-hadoop2.7"
import pyspark
from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession
conf = SparkConf()
sc = SparkContext(conf=conf)
spark = SparkSession.builder.config(conf=conf).getOrCreate() 
import pandas as pd
ip = spark.read.format("csv").option("inferSchema","true").option("header","true").load(r"some other file.csv")
kw = pd.read_csv(r"some file.csv",encoding='ISO-8859-1',index_col=False,error_bad_lines=False)
for i in range(len(kw)):
    rx = '(?i)'+kw.Keywords[i]
    ip = ip.where(~ip['Content'].rlike(rx))
op = ip.toPandas().collect()
op.to_csv(r'something.csv',encoding='utf-8')

toPandas() 之后需要在 PySpark 中执行 collect() 操作才能实现 DataFrame。但是,对于大型数据集不应该这样做,因为toPandas().collect() 会导致数据移动到驱动程序,如果数据集太大而无法放入驱动程序内存,这可能会崩溃。

至于这一行:ip.write.csv('file.csv') 我相信应该将其更改为ip.write.csv('file:///home/your-user-name/file.csv') 以将文件保存在本地 linux 文件系统上,

ip.option("header", "true").csv("file:///C:/out.csv") 将文件保存在本地 Windows 文件系统上(如果您在 Windows 上运行 Spark 和 Hadoop)

ip.write.csv('hdfs:///user/your-user/file.csv') 将文件保存到 HDFS

请告诉我此解决方案是否适合您。

更新

https://github.com/steveloughran/winutils/tree/master/hadoop-2.7.1/bin点击此链接并下载 winutils.exe 文件。在您的 C 盘上创建一个名为 hadoop 的文件夹,并在 hadoop 文件夹内创建另一个名为 bin 的文件夹。将您之前下载的 winutils.exe 放入此目录。 然后您需要编辑系统变量并将变量 HADOOP_HOME 添加到列表中。 完成此操作后,您将不会从 spark 收到 winutils/hadoop 错误。

。 只需在 Windows 搜索中输入“编辑系统环境变量”

【讨论】:

尝试了.collect() 方法。收到此错误:py4j.protocol.Py4JJavaError: An error occurred while calling o86.get. : java.util.NoSuchElementException: spark.sql.execution.pandas.respectSessionTimeZone 你有没有安装 pyarrow ???箭头库是 pyspark 操作 Date 和 TimeStamp 数据所需的默认库。如果未安装,只需通过 pip 安装并尝试再次运行您的脚本 你能添加整个堆栈跟踪吗? 添加了整个回溯。 Stacktrace 显示 Spark 无法找到 winutils.exe 文件。在本地 Windows 机器上运行 spark 时,这是必需的。它实际上让 spark 认为它正在与 hadoop 安装一起交互/运行

以上是关于DF.topandas() - 无法在 hadoop 二进制路径中找到 winutils 二进制文件的主要内容,如果未能解决你的问题,请参考以下文章

Spark DataFrame mapPartitions

07 从RDD创建DataFrame

从RDD创建DataFrame

07 从RDD创建DataFrame

07 从RDD创建DataFrame

07 从RDD创建DataFrame