다른 데이터 프레임의 두 열을 나누기

Question

나는 사용자 로그 파일에 대한 탐색 데이터 분석을 위해 Spark을 사용하고 있습니다. 내가하고있는 분석 중 하나는 호스트 당 일일 평균 요청입니다. 따라서 평균을 계산하려면 DataFrame의 총 요청 열을 DataFrame의 고유 한 요청 열로 나누어야합니다.

total_req_per_day_df = logs_df.select('host',dayofmonth('time').alias('day')).groupby('day').count() 

avg_daily_req_per_host_df = total_req_per_day_df.select("day",(total_req_per_day_df["count"]/daily_hosts_df["count"]).alias("count")) 

이것은 평균을 결정하기 위해 PySpark를 사용하여 작성한 것입니다. daily_hosts_df 및 logs_df이 메모리에 캐시 : 그리고 여기에 내가

AnalysisException: u'resolved attribute(s) count#1993L missing from day#3628,count#3629L in operator !Project [day#3628,(cast(count#3629L as double)/cast(count#1993L as double)) AS count#3630]; 

주를 얻을 오류의 로그입니다. 두 데이터 프레임의 카운트 열을 어떻게 나누나요?

Answer 1

다른 표에서 열을 참조 할 수 없습니다.

from pyspark.sql.functions import col 

(total_req_per_day_df.alias("total") 
    .join(daily_hosts_df.alias("host"), ["day"]) 
    .select(col("day"), (col("total.count")/col("host.count")).alias("count"))) 

Answer 2

이 열 하루에 두 개의 데이터 프레임에 참여하고 카운트 열 하루 비율을 선택하면 데이터를 결합하려면 먼저이 비슷한을 사용 join해야합니다.

total_req_per_day_df = logs_df.select(dayofmonth('time') 
             .alias('day') 
            ).groupBy('day').count() 

avg_daily_req_per_host_df = (
    total_req_per_day_df.join(daily_hosts_df, 
          total_req_per_day_df.day == daily_hosts_df.day 
          ) 
    .select(daily_hosts_df['day'], 
      (total_req_per_day_df['count']/daily_hosts_df['count']) 
      .alias('avg_reqs_per_host_per_day') 
     ) 
    .cache() 
) 

Answer 3

edX 스파크 코스 지정에서 질문입니다. 해결책이 공개 된 이래로 나는 다른, 더 느린 것을 공유하고 그 성능이 향상 될 수 있는지 또는 완전히 스파크가 아닌지 묻는 기회를 갖습니다.

daily_hosts_list = (daily_hosts_df.map(lambda r: (r[0], r[1])).take(30)) 
days_with_hosts, hosts = zip(*daily_hosts_list) 
requests = (total_req_per_day_df.map(lambda r: (r[1])).take(30)) 
average_requests = [(days_with_hosts[n], float(l)) for n, l in enumerate(list(np.array(requests, dtype=float)/np.array(hosts)))] 
avg_daily_req_per_host_df = sqlContext.createDataFrame(average_requests, ('day', 'avg_reqs_per_host_per_day')) 

Answer 4

해결책은 0323 대답을 기준으로하지만 OUTER 조인으로 올바르게 작동합니다.

avg_daily_req_per_host_df = (
    total_req_per_day_df.join(
     daily_hosts_df, daily_hosts_df['day'] == total_req_per_day_df['day'], 'outer' 
).select(
     total_req_per_day_df['day'], 
     (total_req_per_day_df['count']/daily_hosts_df['count']).alias('avg_reqs_per_host_per_day') 
) 
).cache() 

'외부'매개 변수가 없으면 데이터 프레임 중 하나에서 날짜가 누락 된 데이터가 손실되었습니다. 두 데이터 프레임에 같은 날짜가 포함되어 있기 때문에 PySpark Lab2 작업에는 중요하지 않습니다. 하지만 다른 작업에 약간의 통증을 만들 수 있습니다 :)