다음 스파크 df가 있습니다.
다음 명령을 실행하여 Spark를 이미 설치 한 경우에만 로컬에서 실행할 수 있습니다. 그렇지 않으면 Spark 컨텍스트를 자동으로 초기화하는 Databricks 클러스터에 문제를 복제합니다.
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("test").getOrCreate()
sc = spark.sparkContext
spark_dataframe = pd.DataFrame({'id' : ['867', '430', '658', '157', '521', '867', '430', '867'],
'Probability':[0.12, 0.72, 0.32, 0.83, 0.12, 0.49, 0.14, 0.12],
'RAG': ['G', 'R', 'A', 'R', 'G', 'A', 'G', 'G'],
'Timestamp': ['2020-07-01 17-49-32', '2020-07-01 17-49-32', '2020-07-01 17-49-32', '2020-07-01 17-49-32', '2020-07-01 17-49-32', '2020-07-01 16-45-32', '2020-07-01 16-45-32', '2020-07-01 15-45-32']})
spark_dataframe = spark.createDataFrame(spark_dataframe)
이제이 스파크 데이터 프레임을 'id'로 그룹화하고 'RAG'열의 값을 계산하여 다른 열로 나누고 싶습니다. 그러니 이와 같은 것을 얻으십시오.
+---+--------------------+-------------+------------+
| id||G(count)|A(count)|R(count)|Timestamp(max) |
+---+--------------------+-------------+------------+
|867| 2| 1| 0|2020-07-01 17-49-32|
|430| 1| 0| 1|2020-07-01 17-49-32|
|658| 0| 1| 0|2020-07-01 17-49-32|
|157| 0| 0| 1|2020-07-01 17-49-32|
|521| 1| 0| 0|2020-07-01 17-49-32|
+---+--------------------+-------------+------------+
위의 Spark 데이터 프레임을 기반으로 다음과 같은 사전 목록을 만듭니다.
final_list=[]
map_dictionary={"R":0.6, "A":0.3, "G":0.1}
final_list=[{"id": "867", "RAG": "G", "Timestamp": "2020-07-01 17-49-32"}, #because for the id 867 the G column had 2 counts greater than the rest A, R column values on the same row.
{"id": "430", "RAG": "R", "Timestamp": "2020-07-01 17-49-32"} #because G and R had 1 occurrence but R has greater weight based on the map dictionary,...
] #length of the list is equal to 5 since five are the unique rows of the spark df above.
그룹화하고 피벗 할 수 있습니다.
import pyspark.sql.functions as F
#%%
tst = sqlContext.createDataFrame([(867,0.12,'G','2020-07-01 17-49-32'),(430,0.72,'R','2020-07-01 17-49-32'),(658,0.32,'A','2020-07-01 17-49-32'),\
(157,0.83,'R','2020-07-01 17-49-32'),(521,0.12,'G','2020-07-01 17-49-32'),(867,0.49,'A','2020-07-01 16-45-32'),
(430,0.14,'G','2020-07-01 16-45-32'),(867,0.12,'G','2020-07-01 16-45-32')],
schema=['id','Probability','RAG','Timestamp'])
tst1 = tst.groupby('id').pivot('RAG').agg(F.count('Probability').alias('count'),F.max('Timestamp').alias('time_stamp'))
# there will be one time stamp per value of 'RAG'. The below code will find maximum among them
ts_coln = [F.col(x) for x in tst1.columns if 'time_stamp' in x]
tst2 = tst1.withColumn('max_ts',F.greatest(*ts_coln))
결과 :
+ --- + ------- + ------------------- + ------- + --------- ---------- + ------- + ------------
-------+-------------------+
| id|A_count| A_time_stamp|G_count| G_time_stamp|R_count| R_time_stamp| max_ts|
+---+-------+-------------------+-------+-------------------+-------+-------------------+-------------------+
|658| 1|2020-07-01 17-49-32| 0| null| 0| null|2020-07-01 17-49-32|
|430| 0| null| 1|2020-07-01 16-45-32| 1|2020-07-01 17-49-32|2020-07-01 17-49-32|
|521| 0| null| 1|2020-07-01 17-49-32| 0| null|2020-07-01 17-49-32|
|157| 0| null| 0| null| 1|2020-07-01 17-49-32|2020-07-01 17-49-32|
|867| 1|2020-07-01 16-45-32| 2|2020-07-01 17-49-32| 0| null|2020-07-01 17-49-32|
+---+-------+-------------------+-------+-------------------+-------+-------------------+-------------------+
마지막으로 관련없는 열을 삭제할 수 있습니다.
이 기사는 인터넷에서 수집됩니다. 재 인쇄 할 때 출처를 알려주십시오.
침해가 발생한 경우 연락 주시기 바랍니다[email protected] 삭제
몇 마디 만하겠습니다