프로도의 블로그

아래와 같은 방법을 이용하면 날짜를 뺄 수 있다. date_sub(from_unixtime(unix_timestamp('12/03/2010' , 'dd/MM/yyyy')), 10) -- subs 10 days

아래와 같은 방법을 사용하면 복수개의 컬럼에 집계 함수를 사용할 수 있다. from pyspark.sql.functions import countDistinct df.groupBy("col1").agg(countDistinct("col2").alias("col2"), avg("col3").alias("col3"), ...)

spark submit을 하면 spark와 관련된 로그들을 확인할 수 있다. 이 때 너무 많은 로그가 나온다면 필요한 로그를 확인할 수 없다. 보통 정상적으로 수행하고 있는 코드라면 경고, 에러인 데이터만 확인하면 될 것이다. 이럴 때는 아래와 같은 코드를 pyspark 코드 상단에 추가하면 원하는 수준의 로그를 확인할 수 있다. from pyspark.sql import SparkSession spark = SparkSession.builder.\ master('xxx').\ appName('xxx').\ getOrCreate() spark.sparkContext.setLogLevel('WARN') 위에서 보여줄 로그의 수준은 아래 유형중에 선택하면 된다. ALL, DEBUG, ERROR, FATAL,..

아래 코드와 같이 파케 파일을 조회하면 파일 안에 있는 데이터를 확인할 수 있다. data = sqlContext.read.parquet('/.../data.parquet') display(data) # 또는 data.show()

spark dataframe 또는 koalas를 이용해서 DF를 만들고 toPandas()를 이용해서 pandas DF로 변환해야하는 경우가 있다. 필자 같은 경우 DF로 heatmap을 만드는데 koalas DF에서 만들면 에러가 발생해서 pandas df로 변환했다. 문제는 변환 후 조회하면 "index 2 is out of bounds for axis 0 with size" 와 같은 에러가 발생했다. 구체적으로 DF에 NaN 값이 있었고, df.fillna(0)으로 NaN을 0값으로 변환한 경우에 에러가 발생했다. 이경우 toPandas() 코드 윗 부분에 아래와 같은 코드를 추가하면 된다. 파라미터를 -1로 하면 동일한 에러가 발생하는 것을 확인할 수 있다. pd.set_option('displa..

pyspark에서 koalas를 이용해서 DataFrame을 사용하는 작업에서 아래와 같은 에러를 만날 수 있다. Cannot combine the series or dataframe because it comes from a different dataframe. In order to allow this operation, enable 'compute.ops_on_diff_frames' option. 이 경우 에러 메시지에도 나와있는 옵션을 아래와 같이 추가하면 된다. from databricks.koalas.config import set_option, reset_option set_option("compute.ops_on_diff_frames", True) kdf['C'] = kser # Reset ..

spark submit 시 아래와 같은 에러가 발생할 수 있다. 원인은 executor에 할당된 메모리가 부족하다는 의미이다. 이 경우 executor의 할당된 메모리를 늘려주면 된다. cluster.YarnScheduler: Lost executor 20 on xxx: Container killed by YARN for exceeding physical memory limits. 6 GB of 6 GB physical memory used. Consider boosting spark.executor.memoryOverhead. 아래와 같이 spark-submit 시 아래 옵션의 값을 변경해준다. spark-submit --master yarn \ ..... --executor-memory 12g

아래와 같이 초록색 task 박스에 볼드가 없는 경우가 있다. 이런 경우는 아래의 trigger dag 버튼을 눌러서 실행한 경우이다. 배치가 수행된 것이라고 오해하면 안된다.

sprark 배치 수행 중 아래와 같은 에러가 발생했다. 대략적인 원인은 병렬 처리로 인해서 임시 데이터가 삭제된 것이다. Caused by: org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.hdfs.server.namenode.LeaseExpiredException): No lease on .c000.snappy.parquet (inode 4027045532): File does not exist. Holder DFSClient_NONMAPREDUCE_-690256595_53 does not have any open files. hive 테이블에 데이터를 입력하기 전에 alter table truncate partition 하는 부분이 있었는데 그 부..

아래와 같이 한 번의 쿼리로 여러 파티션이 데이터를 입력할 수 있다. FROM table_name INSERT OVERWRITE TABLE table_1 PARTITION (part_date = '20200101', cust_type = '1') select * where table_name.part_date = '20200101' and table_name.cust_type = '1' INSERT OVERWRITE TABLE table_1 PARTITION (part_date = '20200101', cust_type = '2') select * where table_name.part_date = 'v' and table_name.cust_type = '2' INSERT OVERWRITE TABLE t..