Я запускаю PySpark в Jupyter Notebook, пытаясь загрузить много больших файлов JSON.Ранее я тестировал тот же сценарий только с одним файлом (размером 11 ГБ), и он работал без проблем.
Мой сценарий Python:
import os, sys, pandas, time
import findspark
findspark.init('/opt/cloudera/parcels/SPARK2/lib/spark2')
import pyspark
os.environ['PYSPARK_PYTHON'] = "/opt/cloudera/Anaconda3/bin/python"
from pyspark import SparkConf, SparkContext, StorageLevel
from pyspark.sql.types import *
from pyspark.sql import Row, SparkSession, HiveContext, SQLContext
target_directory = "removed for this post"
top_directory = "removed for this post"
sc_conf = SparkConf()
sc_conf.setAppName("test"),
sc_conf.set('spark.ui.port', 49051)
sc_conf.set('spark.executor.memory', '18g'),
sc_conf.set('spark.executor.cores', '4')
sc_conf.set('spark.executor.extraJavaOptions', '-XX:+UseG1GC'),
sc_conf.set('spark.driver.memory', '18g'),
sc_conf.set('spark.yarn.am.memory', '8g'),
sc_conf.set('spark.yarn.am.cores', '4'),
sc_conf.set('spark.task.cpus','1'),
sc_conf.set('spark.serializer','org.apache.spark.serializer.KryoSerializer'),
ses = (
SparkSession
.builder
.config(conf=sc_conf)
.enableHiveSupport()
.getOrCreate()
)
sc = ses.sparkContext
start_time = time.time()
sqlContext = HiveContext(ses)
print("LOADING JSON DF")
start_time_json = time.time()
jsonDF = sqlContext.read.json(top_directory + "traceroute*")
print("SUCCESS")
print("Time elapsed: " + str(time.time() - start_time_json) + " seconds")
print("\n")
jsonDF.printSchema()
jsonDF.show(3)
ses.sql("USE my_table_name")
start_time_orc = time.time()
print("WRITING ORC")
#jsonDF.write.format("orc").saveAsTable("main_orc.json")
jsonDF.write.orc("../traceroute_orc")
print("SUCCESS")
print("Time elapsed: " + str(time.time() - start_time_orc) + " seconds")
#print(time.time() - start_time_orc)
print("\n")
start_time_parquet = time.time()
print("WRITING PARQUET")
jsonDF.write.parquet("../traceroute_parquet")
print("SUCCESS")
print("Time elapsed: " + str(time.time() - start_time_parquet) + " seconds")
print("\n")
print("Total time elapsed: " + str(time.time() - start_time) + " seconds")
sc.stop()
Затем я получаю эту ошибку:
---------------------------------------------------------------------------
Py4JJavaError Traceback (most recent call last)
/opt/cloudera/parcels/SPARK2/lib/spark2/python/pyspark/sql/utils.py in deco(*a, **kw)
62 try:
---> 63 return f(*a, **kw)
64 except py4j.protocol.Py4JJavaError as e:
/opt/cloudera/parcels/SPARK2/lib/spark2/python/lib/py4j-0.10.6-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
319 "An error occurred while calling {0}{1}{2}.\n".
--> 320 format(target_id, ".", name), value)
321 else:
Py4JJavaError: An error occurred while calling o6192.json.
: org.apache.spark.sql.AnalysisException: Found duplicate column(s) in the data schema: `result`, `lts`, `paris_id`;
at org.apache.spark.sql.util.SchemaUtils$.checkColumnNameDuplication(SchemaUtils.scala:85)
at org.apache.spark.sql.util.SchemaUtils$.checkColumnNameDuplication(SchemaUtils.scala:67)
at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:421)
at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:239)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:227)
at org.apache.spark.sql.DataFrameReader.json(DataFrameReader.scala:397)
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:282)
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)
During handling of the above exception, another exception occurred:
AnalysisException Traceback (most recent call last)
<ipython-input-21-99870ac0546e> in <module>()
67 print("LOADING JSON DF")
68 start_time_json = time.time()
---> 69 jsonDF = sqlContext.read.json(top_directory + "traceroute*")
70 print("SUCCESS")
71 print("Time elapsed: " + str(time.time() - start_time_json) + " seconds")
/opt/cloudera/parcels/SPARK2/lib/spark2/python/pyspark/sql/readwriter.py in json(self, path, schema, primitivesAsString, prefersDecimal, allowComments, allowUnquotedFieldNames, allowSingleQuotes, allowNumericLeadingZero, allowBackslashEscapingAnyCharacter, mode, columnNameOfCorruptRecord, dateFormat, timestampFormat, multiLine, allowUnquotedControlChars)
259 path = [path]
260 if type(path) == list:
--> 261 return self._df(self._jreader.json(self._spark._sc._jvm.PythonUtils.toSeq(path)))
262 elif isinstance(path, RDD):
263 def func(iterator):
/opt/cloudera/parcels/SPARK2/lib/spark2/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py in __call__(self, *args)
1158 answer = self.gateway_client.send_command(command)
1159 return_value = get_return_value(
-> 1160 answer, self.gateway_client, self.target_id, self.name)
1161
1162 for temp_arg in temp_args:
/opt/cloudera/parcels/SPARK2/lib/spark2/python/pyspark/sql/utils.py in deco(*a, **kw)
67 e.java_exception.getStackTrace()))
68 if s.startswith('org.apache.spark.sql.AnalysisException: '):
---> 69 raise AnalysisException(s.split(': ', 1)[1], stackTrace)
70 if s.startswith('org.apache.spark.sql.catalyst.analysis'):
71 raise AnalysisException(s.split(': ', 1)[1], stackTrace)
AnalysisException: 'Found duplicate column(s) in the data schema: `result`, `lts`, `paris_id`;'
Вот схема JSON одного объекта: https://pastebin.com/W6nHuJuL
Почему в этих столбцах есть дубликаты?Как вы можете видеть из JSON, который я опубликовал, дубликатов нет.