프로도의 블로그

어떤 데이터를 추출할 때에는 월요일~일요일 사이의 데이터를 가져와야 할 경우가 있다. 예를들어 주간 실적같은 것을 보여줄 때가 그렇다. 이럴 때는 월요일~일요일의 일자를 알아야 한다. 월요일만 알면 6일을 더하면 일요일이 되기 때문에 월요일 날짜를 찾는게 집중하면 된다. hive에서는 아래와 같이 쿼리를 실행하면 월요일의 일자를 알 수 있다. select next_day(date_sub('2019-01-01', 7), 'MON') hive 1.2 버전 이상에서만 될 수 있으니 hive의 버전을 확인하자. 참고 stackoverflow.com/questions/33196651/how-to-get-the-date-of-the-first-day-of-a-week-given-a-time-stamp-in-hado..

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..

spark에서 createDataFrame을 하면 df를 만들 수 있다. 이 때 스키마를 지정해야하는데 기존의 df와 동일하게 만들고 싶다면 df.schema를 바로 이용할 수 있다. list_rslt_on_driver_node = df_rslt.collect() df_rslt_on_driver_node = spark.createDataFrame(list_rslt_on_driver_node,schema=df_rslt.schema) df_rslt_on_driver_node.repartions(1) df2 = df_rslt_on_driver_node.repartition(1)

spark에서 dataframe을 이용할 때 데이터는 worker node에 분산되어 저장된다. 이 때 어떠한 필요에 의해서 데이터를 master node로 보내야 하는 경우가 있다면 collect()를 이용하면 된다. 필자의 경우 df의 데이터를 oracle 데이터로 보내기 위해서 이 작업이 필요했다. master node는 오라클 db에 접근이 가능하지만 worker node는 보안 정책으로 인해 접근할 수 없었다. 그래서 데이터를 master node로 보내고 master node의 데이터를 다른 db로 보내려고 했다. 결론적으로는 실패 해서 다른 방법을 이용했지만 어쨌든 데이터를 master node에 모으고 싶다면 이 방법을 이용하면 된다. df.collect() 데이터가 많은 경우에는 데이터가..

만들어 놓은 데이터프레임이 몇개로 분할되어 있는지 확인하고 싶을 때는 아래와 같이 실행하면 파티션 수를 확인할 수 있다. df.rdd.getNumPartitions() 추가로 파티션을 조정하고 싶다면 repartition() 이나 Coalesce()를 사용할 수 있다. df = df.repartition(5) 또는 df = df.coalesce(5) 참고 sparkbyexamples.com/pyspark/pyspark-repartition-vs-coalesce/

airflow를 이용해서 data pipeline을 관리하고 있다면 신규로 추가 되는 작업에 대해서 테스트가 필요하다. 이 때 dag를 기준으로 테스트를 하면 이전에 처리되는 task들을 모두 실행해야하기 때문에 시간과 리소스에 낭비가 발생한다. 이럴 때는 특정 task만 실행하는 방법을 사용하면된다. #airflow dag명 task명 execution_date airflow dag task 2020-01-11 참고 airflow.apache.org/docs/stable/tutorial.html

데이터프레임으로 테이블을 만들어야 하는 경우에는 아래 같이 코드를 작성하면 된다. 테이블을 생성하고 데이터를 넣기 df.write.mode("overwrite").saveAsTable("스키마.테이블명") # 데이터 조회하기 spark.sql("select * from 스키마.테이블명").show() 기존에 존재하는 테이블에 데이터만 넣기 df.write.mode("append").saveAsTable("스키마.테이블명") spark.sql("select * from 스키마.테이블명").show() 참고 kontext.tech/column/spark/294/spark-save-dataframe-to-hive-table

spark 2.3에서 잘 수행되던 코드가 spark 3.0에서 아래와 같은 오류가 발행했다. select unix_timestamp(update_dt) 오류 메세지 Caused by: java.time.format.DateTimeParseException: Text '2020-09-16 16:24:08.0' could not be parsed, unparsed text found at index 19 at java.time.format.DateTimeFormatter.parseResolved0(DateTimeFormatter.java:1952) at java.time.format.DateTimeFormatter.parse(DateTimeFormatter.java:1777) at org.apache.spa..

여러가지 이유로 hive 테이블에 partition을 확인해야 하는 경우가 있다. 이런 경우 아래와 같이 조회를 하면 hive table의 partition을 확인할 수 있다. show partitions

linux 또는 맥북에서 curl을 이용하는 방법과 윈도우에서 실행하는 방법이 다르다. 윈도우에서 json 파일을 호출하려면 아래와 같은 방법을 이용해야 한다. $json = Get-Content ‪D:\workplace\projects\druid_create_datasource.json -Raw curl -Method POST -Header $Header -ContentType "application/json" -Body $json uri http://druid-server:/druid/indexer/v1/task