🐥

스파크에서 Join 시에는 상황에 따라 Shuffle 연산을 수행해야 하고, 이는 executor 사이의 방대한 데이터 이동을 야기한다. 스파크에서 사용되는 조인의 방식은 아래와 같다. 1. Broadcast Hash Join 두 개의 데이터를 조인할 때 한 쪽이 매우 작고 한 쪽은 매우 큰 사이즈의 데이터일 때, 더 작은 쪽의 데이터가 Driver에 의해 모든 executor로 복사되는 방식이다. 사이즈가 작은 테이블의 키 값을 해시 테이블로 만든다. 해시 테이블의 값을 큰 테이블 파티션에 복사해준다. 각 파티션 내부에서 조인을 진행한다. 이 방식은 shuffle을 생략할 수 있기 때문에 데이터 이동에 따른 코스트가 사라진다. 모든 executor에 데이터를 복사하기 때문에 복사하려는 dataset이 ..

2.4.4 이하와 2.4.5 이상 버전에서 하둡 configuration 설정하는게 좀 다른 듯 하다. Spark 버전 2.4.4 이하 spark = SparkSession.builder.appName("myapp") \ .config("some.config", "some.value") \ .getOrCreate() # signature V4 를 사용하는 리전일 경우 아래 property 설정 spark.sparkContext.setSystemProperty("com.amazonaws.services.s3.enableV4", "true") # s3 정보 설정 spark.sparkContext._jsc.hadoopConfiguration().set(f"fs.s3a.bucket.{mybucket}.endpo..

sparksession을 사용하여 가져올 수 있다. (Spark 2.0 이상) import pyspark from pyspark.sql import SparkSession spark = SparkSession.builder.appName("s3test").config("spark.some.config.option", True).getOrCreate() df = spark.read.parquet("s3a://my-bucket/my/s3/path") schema, option등 HiveContext를 이용하여 hdfs 파일을 로드할 때와 동일하게 옵션도 줄 수 있음 참고: https://stackoverflow.com/questions/44629156/how-to-read-parquet-data-from-s..

뮨제상황 dataframe 두 개를 조인하려는데 아래와 같은 워닝이 주르륵 발생하더니 오류를 출력하고 어플리케이션이 종료됐다. ... 23/05/22 04:39:35 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again. 23/05/22 04:39:35 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again. 23/05/22 04:39:35 WARN TaskMemoryManager: Failed to allocate a page (2097152 bytes), try again. 23/05/22 04:39:36 WARN TaskMemoryMa..

data_3 = hc.read.\ csv('/my/path/partition={PARTITION}/*'.format(PARTITION=my_partition), header=False, schema=customSchema) # header가 있을 경우 schema는 따로 없어도 됨 (header=True)