question Re: Error spark job input is too large to fit in a byte array in Support Questions

Error spark job input is too large to fit in a byte array

Yosieam — Thu, 19 May 2022 02:35:41 GMT

I found my error on spark job and i can't solve this error yet until i got stuck, this is the error i got

22/05/19 09:32:40 ERROR util.SparkUncaughtExceptionHandler: Uncaught exception in thread Thread[Executor task launch worker for task 1,5,main]

java.lang.OutOfMemoryError: input is too large to fit in a byte array

at org.spark_project.guava.io.ByteStreams.toByteArrayInternal(ByteStreams.java:194)

at org.spark_project.guava.io.ByteStreams.toByteArray(ByteStreams.java:220)

at org.apache.spark.input.WholeTextFileRecordReader.nextKeyValue(WholeTextFileRecordReader.scala:79)

at org.apache.spark.sql.execution.datasources.RecordReaderIterator.hasNext(RecordReaderIterator.scala:39)

at org.apache.spark.sql.execution.datasources.HadoopFileWholeTextReader.hasNext(HadoopFileWholeTextReader.scala:52)

at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)

at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:101)

at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.nextIterator(FileScanRDD.scala:181)

at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:101)

at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)

at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)

at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:645)

at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:265)

at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:257)

at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:858)

at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)

at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:346)

at org.apache.spark.rdd.RDD.iterator(RDD.scala:310)

at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)

at org.apache.spark.scheduler.Task.run(Task.scala:123)

at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)

at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1289)

at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)

at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)

at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)

at java.lang.Thread.run(Thread.java:748)

I want to process log file which has 20gb size and save it to hive table and this is my code:

spark = SparkSession \

.builder \

.config("spark.rpc.message.maxSize", "1024") \

.appName("Spark Client") \

.getOrCreate()

filepath = "/hdlog/dbgen2log/postgresql_2021-09-14_Tue.log"

# files = spark.read.text(filepath, wholetext = True)

# getData = files.select("value").collect()

dataGenerator = (i for i in spark.read.text(filepath, wholetext = True).select("value").collect()[0].value.split('time='))

tanggal, pid, db, user, client, app, line, kategori, query = [

], [], [], [], [], [], [], [], []

for i in dataGenerator:

if i == "":

continue

else:

def getData():

stat = re.findall(r"LOG: statement:", i)

aud = re.findall(r"LOG: AUDIT:", i)

execu = re.findall(r"LOG: execute", i)

if stat != []:

return "statement"

elif aud != []:

return "AUDIT"

elif execu != []:

return "execute"

else:

False

def getKategori(value):

if value.find("LOG:statement:") > -1:

return value[value.find("LOG:statement:") + 4:value.find("nt:") + 2]

elif value.find("LOG:AUDIT") > -1:

return value[value.find("LOG:AUDIT") + 4:value.find(":S")]

elif value.find("LOG:execute") > -1:

return value[value.find("LOG:execute") + 4:value.find("ute") + 3]

else:

pass

def getQuery(value):

if value == "statement":

query.append(ReSpace[5].replace(

" line=" + line[-1] + " LOG: statement: ", "") + ",".join(ReSpace[6:]))

elif value == "execute":

query.append(ReSpace[5].replace(

" line="+line[-1]+" LOG: execute ", "") + ",".join(ReSpace[6:]))

else:

query.append(",".join(ReSpace[6:]))

if getData():

files = i

ReSymbol1 = re.sub(r'\t', ' ', files)

ReSymbol2 = re.sub(r'\n', ' ', ReSymbol1)

ReSpace = re.sub(r' +', ' ', ReSymbol2).split(",")

tanggal.append(ReSpace[0].replace("WIB", ""))

pid.append(ReSpace[1].split(" ")[1][4:])

db.append(ReSpace[1].split(" ")[2][3:])

user.append(ReSpace[2].replace(" ", "")[4:])

client.append(ReSpace[3].replace(" ", "")[7:])

app.append(ReSpace[4].replace(" ", "")[4:])

line.append(ReSpace[5].replace(" ", "")[

5:ReSpace[5].replace(" ", "").find("LOG")])

kategori.append(getKategori(str(ReSpace[5].replace(" ", ""))))

getQuery(getData())

else:

pass

df = pd.DataFrame({

'tanggal': tanggal,

'pid': pid,

'db': db,

'user': user,

'client': client,

'app': app,

'line': line,

'kategori': kategori,

'query': query

})

df = df[["tanggal", "pid", "db", "user",

"client", "app", "line", "kategori", "query"]]

datas = spark.createDataFrame(df)

spark.sql("use log_analytics")

datas.registerTempTable("table_dbgen2log")

datas.write.format("parquet").mode(

"Overwrite").saveAsTable("test_dbgen2_log_query")

Re: Error spark job input is too large to fit in a byte array

mszurap — Wed, 22 Jun 2022 09:45:52 GMT

Hi @Yosieam , Using "collect" method is not recommended as it needs to collect the data to the Spark driver side and as such it needs to fit the whole dataset into the Driver's memory. Please rewrite your code to avoid the "collect" method.

Re: Error spark job input is too large to fit in a byte array

Yosieam — Thu, 23 Jun 2022 02:34:32 GMT

i want to collect them because i want to make them as 1 string and split them to array and store as RDD, because my log is need to split into several pieces with seperator i give

Re: Error spark job input is too large to fit in a byte array

Yosieam — Thu, 23 Jun 2022 03:59:41 GMT

I have change my code not using collect, but I use wholeTextFiles() and use flatMap() for split the string , but i have difference error

java.lang.OutOfMemoryError: input is too large to fit in a byte array

at org.spark_project.guava.io.ByteStreams.toByteArrayInternal(ByteStreams.java:194)

at org.spark_project.guava.io.ByteStreams.toByteArray(ByteStreams.java:220)

at org.apache.spark.input.WholeTextFileRecordReader.nextKeyValue(WholeTextFileRecordReader.s

cala:79)

Re: Error spark job input is too large to fit in a byte array

mszurap — Thu, 23 Jun 2022 08:47:26 GMT

wholeTextFiles is also not a scalable solution.
https://spark.apache.org/docs/3.1.3/api/python/reference/api/pyspark.SparkContext.wholeTextFiles.html

"Small files are preferred, as each file will be loaded fully in memory."

Re: Error spark job input is too large to fit in a byte array

Yosieam — Fri, 24 Jun 2022 07:32:16 GMT

Now I am facing the error like this

Error Exception in thread "dispatcher-event-loop-40" java.lang.OutOfMemoryError: Requested array size exceeds VM limit

And I stuck with this one for about a month, everty time I try to run the job its always stucked there, i have increased the driver and executor memory but it didnt do nothing

Re: Error spark job input is too large to fit in a byte array

mszurap — Fri, 24 Jun 2022 07:36:47 GMT

Hi,

The "Requested array size exceeds VM limit" means that your code tries to instantiate an array which has more than 2^31-1 elements (~2 billion) which is the max size of an array in Java. You cannot solve this with adding more memory. You need to split the work between executors and not process data on a single JVM (Driver side).

Re: Error spark job input is too large to fit in a byte array

Yosieam — Fri, 24 Jun 2022 07:54:35 GMT

Can you tell me where I should check that the process data in single JVM? the purpose of my spark job is writing the result to hive table and the oom comes when the job try to write all data into hive table

Re: Error spark job input is too large to fit in a byte array

RangaReddy — Thu, 01 Sep 2022 04:20:31 GMT

Hi @Yosieam

Thanks for sharing the code. You forgot to share the spark-submit/pyspark command. Please check what is executor/driver memory is passed to the spark-submit. Could you please confirm file is in local system/hdfs system.