프로도의 블로그

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

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 하는 부분이 있었는데 그 부..

스파크 잡을 실행하면 아래와 같은 에러 메지시가 나타날 때가 있다. Service 'SparkUI' failed after 16 retries (starting from 4040)! Consider explicitly setting the appropriate port for the service 'SparkUI' (for example spark.ui.port for SparkUI) to an available port or increasing spark.port.maxRetries. 원인은 동시에 어러개의 잡이 실행될 때 SparkUI를 생성하기 위한 포트가 이미 사용 중이면 발생한다. 포트를 찾아보는 기본 값은 16이다. 가장 간단한 해결 방법은 더 많은 포트를 찾아보고 비어있는 포트에서 Spark..

pyspark에서 oracle로 데이터를 넣을 때 아래과 같은 에러가 발생할 때가 있다. ORA-01861 : 리터럴이 형식 문자열과 일치하지 않음 직접적인 원인은 spark의 데이터 타입과 오라클의 데이터 타입이 일치하지 않는 경우에 발생한다. 필자와 같은 경우에는 날짜타입의 컬럼일 때 위와 같은 에러가 발생했다. 이 문제를 해결하기 위해서는 spark에서 data frame의 컬럼의 데이터 타입을 변경해야 한다. 수정 전에 DF의 컬럼 타입을 보면 아래와 같다. DataFrame[visit_dt : string] spark에서 날짜타입이 string일 때 timestamp로 변경하면 oracle의 컬럼이 date 타입일 때 문제가 해결된다. 데이터 타입을 수정하는 방법은 아래와 같다. df_rslt ..

pandas는 spark에서 분산 병렬 처리가 되지 않기 때문에 대용량 데이터를 다루기에는 한계가 있다. 그렇다고 spark의 dataframe을 이용하면 pandas에 비해서 기능이 부족하거나 불편한 경우가 있다. 이런 경우에는 databricks에서 만든 koalas를 이용하면 된다. koalas는 분산 병렬처리가 가능하고, 문법도 pandas와 유사해서 어려움 없이 이용할 수 있다. 아래는 koalas를 이용해 df의 describe() 함수를 이용하는 방법이다. import databricks.koalas as ks sdf = spark.sql("select cnt from table") # koalas df로 변환 kdf = sdf.to_koalas() kdf.describe() ##결과 cou..

spark에서 df를 조인한 후에 원하는 컬럼을 선택해야 한다. 이때 sql처럼 편하게 "*" 기호를 쓰거나, 원하는 컬럼을 선택하기를 원한다. 이 때는 아래와 같이 하면된다. "*"를 쓰기 위해서는 alias로 df의 별칭을 지정해줘야 한다. df = (df_a.alias("a").join(df_b.alias("b"), df_a.prd_no == df_b.prd_no) .selectExpr("a.*", "b.mem_no"))

spark를 실행 중에 config 값을 변경해야하는 경우가 있다. 필자 같은 경우에는 처음 spark submit을 할 때는 spark.master를 yarn을 사용하다가 중간에 local mode로 변경해야 하는 경우가 있다. 조금 더 자세히 이야기를 하자면 yarn cluster에서는 필자가 접근하려는 db에 방화벽이 있기 때문에 접속을 못하고 driver node에서만 접근이 가능했기 때문에 사용했다. # spark session을 stand alone node로 변경 conf = spark.sparkContext._conf.setAll([('spark.master', 'local[10]'),('spark.driver.memory','8g')]) spark.sparkContext.stop() sp..

아래와 같은 패턴을 이용하면 문자열로 날짜 값을 만들 수 있다. select from_unixtime(unix_timestamp('20161023235959', 'yyyyMMddHHmmss'), 'yyyy-MM-dd HH:mm:ss') as new_format

spark dataframe에서 조인을 하면 동일한 컬럼이 2개 생길 수 있다. 안 생길 때도 있다. 어쨌든 동일한 컬럼이 중복으로 생기면 제거해줘야 한다. 이 때는 2가지 방법이 있다. .select()를 이용해 사용할 컬럼만 선택하거나 .drop을 이용해서 필요없는 컬럼을 제거해야 한다. 여러 개의 컬럼을 선택하거나 삭제할 때는 아래와 같은 방법을 이용해야 한다. df = (df.join(df_b, (df.mem == df_b.mem) & (df.prd == df_b.prd), "left_anti") .drop("df_b.mem_no,df_b.prd_no")) spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=join pyspark..