from pyspark import SparkContext
from pyspark import SparkConf
from pyspark.sql import *

def check_csv_to_parquet(parquet_s3_location,csv_s3_location):
    
    parquet_df=spark.read.parquet(parquet_s3_location)
    print("parquet file 전체 row 수 : ", parquet_df.count())
    print("parquet file 전체 column : ", parquet_df.columns) # 리스트형태임
    
    csv_df=spark.read.csv(csv_s3_location,header=True,inferSchema='false',multiLine=True,\
                          quote='"',escape='\\',sep=',',ignoreLeadingWhiteSpace='true',\
                          ignoreTrailingWhiteSpace='true',mode='PERMISSIVE',encoding="UTF-8")
    print("csv file 전체 row 수 : ", csv_df.count())
    print("csv file 전체 column : ", csv_df.columns)
    
    compare_list=csv_df.columns
    for col_name in compare_list:
        if col_name not in parquet_df.columns:
            compare_list.remove(col_name)
    
    result = parquet_df.exceptAll(csv_df).orderBy(compare_list).count()==0
    
    if result is False:
        print("컬럼 정합성 이상없음")
    else:
        print("서로 컬럼이 다른 내용이 있음. 확인필요.")
        print("해당리스트 : ",list(set(csv_df.columns).difference(set(parquet_df.columns))))
    
    return None

check_csv_to_parquet("s3a://my_bucket/my_parquet","s3a://my_bucket/UTF8_my_csv.csv")