[Spark] pyspark 3.0 dataframe new function 테스트#2(csv관련)

 

2020/07/03 - [Data Science/Spark] - [Spark] spark 3.0 dataframe 새로운 기능 정리 #1

직전 포스트에서 준비한 데이터로 신규기능, 개선기능을 테스트

○ 테스트데이터 확인

○ csv관련 ( from_csv, to_csv, schema_of_csv )
from_json의 csv(comma seperated value)대응기능, 컴마(',')로 구분된 데이터를 다룰때 사용

- to_csv(col, options={})
col : list형식으로 된 컬럼 [a,b,c,d,1,2,3]
options : spark.read.csv에서 사용하는 option들을 사용할 수 있다. 자세한 옵션들은 아래 url참고
https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.DataFrameReader.csv

#to_csv 테스트
data = df.head(df.count()) 
data_list = []
i = 1
for d in data : 
  tmp = [i, d]
  data_list.append(tmp)
  i += 1
to_csv_test_df_tmp = spark.createDataFrame(data_list, ("key", "value"))
to_csv_test_df = to_csv_test_df_tmp.select(func.to_csv(to_csv_test_df_tmp.value).alias("csv_format"))
to_csv_test_df.show()

- from_json (col, schema, options={})
col : csv포맷의 string column
schema : DDL format
options : spark.read.csv에서 사용하는 option들을 사용할 수 있다. 자세한 옵션들은 아래 url참고
https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.DataFrameReader.csv

#from_csv테스트
#schema는 structType이 아닌 string으로 해줘야함.
str_schema = """year string, 
                month string, 
                geo int, 
                nongeo int, 
                etc int, 
                fail int,
                sum int
                """
to_csv_test_df.select(func.from_csv("csv_format", str_schema)).show()
#list로 반환한다.

 

- schema_of_csv(csv, options={})
이렇게 csv schema를 지정하는 것은 데이터를 모두 알고 있을때 가능한 일이다.
따라서 대략적인 csv schema를 알고자 할 때, schema를 모두 쓰는게 귀찮을 때 schema_of_csv를 사용한다.

# schema_of_csv 테스트
csv_list = [row.csv_format for row in to_csv_test_df.head(1)]
csv_str = ','.join(csv_list)
print("csv_str : " + csv_str)

to_csv_test_df.select(func.schema_of_csv(csv_str)).collect()

# dataframe안에서 시도 
# 1. csv_format 컬럼
# 2. csv_format 컬럼의 csv schema infer 시도
# 3. csv_format 컬럼의 csv schema infer options넣고 시도
to_csv_test_df.select("csv_format", func.split("csv_format",','),func.schema_of_csv("csv_format"),
                      func.schema_of_csv(csv="csv_format", options={'sep':','})).show()
                      
#결과는 잘 안된다.

 

 

Reference

http://spark.apache.org/releases/spark-release-3-0-0.html

Spark Release 3.0.0 | Apache Spark

https://medium.com/javarevisited/spark-3-0-new-functions-in-a-nutshell-a929fca93413

Spark 3.0  — New Functions in a Nutshell