Haonly's Blog

여러가지 이유로 Spark DataFrame 의 row count 보다는 크기를 계산하여야 한다.

• Broadcast join 이 적합한지 확인용
• Executor 자원을 얼마나 할당할지 확인용
• 데이터 크기를 비교하여 DataFrame 이 정상적으로 생긴 것인지 확인용
• 다양하게...

disk 에 저장하지 않고도 Spark 에서 DataFrame 의 크기를 알려주는 API 가 있다.

import org.apache.spark.util.SizeEstimator
// dataframe 생성
SizeEstimator.estimate(df)

이런 방법으로 데이터 크기를 측정할 수 있다.

Spark Dataframe 에서 칼럼 사이즈 구하기: 스칼라 & 파이썬

1. Scala

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

// 스파크 세션 생성
val spark = SparkSession.builder()
  .appName("ArrayLengthExample")
  .getOrCreate()

// 예제 데이터프레임 생성
import spark.implicits._
val df = Seq(
  (Array(1.0, 2.0, 3.0)),
  (Array(4.0, 5.0)),
  (Array.empty[Double])
).toDF("double_array")

// double_array 배열의 길이를 계산
val dfWithArrayLength = df.withColumn(
  "array_length",
  size(col("double_array"))
)

// 결과 출력
dfWithArrayLength.show(truncate = false)

2. Python

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, size

# 스파크 세션 생성
spark = SparkSession.builder \
    .appName("ArrayLengthExample") \
    .getOrCreate()

# 예제 데이터프레임 생성
data = [
    ([1.0, 2.0, 3.0],),
    ([4.0, 5.0],),
    ([],)
]

df = spark.createDataFrame(data, ["double_array"])

# double_array 배열의 길이를 계산
df_with_array_length = df.withColumn(
    "array_length",
    size(col("double_array"))
)

# 결과 출력
df_with_array_length.show(truncate=False)

오류

error: scala.reflect.internal.MissingRequirementError: object java.lang.Object in compiler mirror not found.

원인

scala 2.11  버전과 java SDK 버전이 맞지 않아 발생하는 오류

해결방법

scala 버전에 맞게 java SDK를 8 버전으로 바꾼다.

나의 경우 9버전이었다.

에러 메시지

Caused by: java.lang.AssertionError: assertion failed: Concurrent update to the commit log. Multiple streaming jobs detected for

원인

로그에 동시 업데이트 하기 때문에 발생

스파크 스트리밍에서 동일한 체크포인트 위치에 두 개 이상의 다른 스파크 작업이 업데이트 하려 할 때 발생한다.

이 때 스파크 설정의 checkpointLocation을 다른 위치로 사용하면 해결할 수 있다.

두 개의 다른 스파크 스트리밍 작업에서 checkpoint 위치를 다르게 한다.

      .option("checkpointLocation", checkpointPath1)

      .option("checkpointLocation", checkpointPath2)

Thread 몇개로 돌리는 애플리케이션이 있다.

어떤 thread가 어떤 작업을 하는지 알고싶었다. 그래서 thread id 찍어보고 싶은데 어떻게 찍는지 몰라서 찾아봄

val threadId = Thread.currentThread.getId

생각보다 많이 간단...!

문자열 split 방법

val s = "aa bb cc"
val splited = s.split(" ")

일반적인 split은 위와 같이 하면 된다.


dot(.)이 포함된 문자열을 split 하려고 했는데 안된다.
아래와 같이 하면 된다.

val s = "aa.bb.cc"
val splited = s.split("\\.")

dot(.)은 자바 정규식 예약어이기 때문에 \(back slash)가 필요한데 \ 자체도 예약어이기 때문에 \\ 두번 붙여야 한다.