인프런 커뮤니티 질문&답변

질문

1. 만약 문제에서 카운트를 출력하라는 조건이 없었으면 persist를 호출 하지않아도 같은 퍼포먼스가 나오나요? (문제: ... 각각 데이터를 로드한 후 카운트를 출력하고,...)

   
   

   count 함수를 제외하면 action 계열 함수가 없는데 count 함수가 없는 경우에도 persist가 유의미 한지 궁금합니다.

    

2. [SQL/DataFrame] - [Details for Query0] 순서도(마지막에 첨부) 에서 Scan csv 가 3번 나타나는 이유는 무엇인가요??
   csv 파일을 읽어오는 함수는 두번 실행했는데 왜 3번 실행되는지 궁금합니다.

    

참고

• 목차

  • 질문-참고-코드-순서도

• 코드 구조

1. HDFS 경로 정의

2. 스키마 정의

3. 데이터프레임 생성

4. time_recorded에 의한 중복열 제거

5. employee_count 컬럼 추가를 위한 join

6. IT Services and IT Consulting 회사 추출

7. 두 테이블 결합

8. 결과 출력

• dropDuplicate가 정확히 어떤 행을 drop 하는지 몰라서 egg-max 조합으로 중복 행을 제거함

코드

"""
>>> company_industries_df.printSchema()
root
 |-- company_id: string (nullable = true)
 |-- industry: string (nullable = true)

>>> company_industries_df.show(5)
+----------+--------------------+
|company_id|            industry|
+----------+--------------------+
|    391906|Book and Periodic...|
|  22292832|        Construction|
|     20300|             Banking|
|   3570660|Book and Periodic...|
|    878353|Staffing and Recr...|
+----------+--------------------+
only showing top 5 rows

>>> employee_counts_df.printSchema()
root
 |-- company_id: string (nullable = true)
 |-- employee_count: string (nullable = true)
 |-- follower_count: string (nullable = true)
 |-- time_recorded: string (nullable = true)

>>> employee_counts_df.show(5)
+----------+--------------+--------------+-------------+
|company_id|employee_count|follower_count|time_recorded|
+----------+--------------+--------------+-------------+
|    391906|           186|         32508|   1712346173|
|  22292832|           311|          4471|   1712346173|
|     20300|          1053|          6554|   1712346173|
|   3570660|           383|         35241|   1712346173|
|    878353|            52|         26397|   1712346173|
+----------+--------------+--------------+-------------+
only showing top 5 rows


+----------+--------------------+--------------+--------------+-------------+
|company_id|            industry|employee_count|follower_count|time_recorded|
+----------+--------------------+--------------+--------------+-------------+
|      1353|IT Services and I...|        596046|      14829798|   1713536641|
|      1353|IT Services and I...|        595935|      14822564|   1713448937|
|      1353|IT Services and I...|        595844|      14817360|   1713407024|
|      1353|IT Services and I...|        595782|      14807439|   1713279321|
|      1353|IT Services and I...|        595669|      14800204|   1713208435|
+----------+--------------------+--------------+--------------+-------------+
only showing top 5 rows

"""

import time
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
from pyspark.sql.functions import col, max

spark = SparkSession.builder.getOrCreate()

# 1. 파일 경로 정의
company_industries_path = (
    "/home/spark/sample/linkedin_jobs/companies/company_industries.csv"
)
employee_counts_path = "/home/spark/sample/linkedin_jobs/companies/employee_counts.csv"
# 2. 스키마 정의
company_industries_schema = StructType(
    [
        StructField("company_id", IntegerType(), True),
        StructField("industry", StringType(), True),
    ]
)
employee_counts_schema = StructType(
    [
        StructField("company_id", IntegerType(), True),
        StructField("employee_count", IntegerType(), True),
        StructField("follower_count", IntegerType(), True),
        StructField("time_recorded", IntegerType(), True),
    ]
)

# 3. 데이터프레임 생성
company_industries_df = (
    spark.read.option("header", "true")
    .option("multiLine", "true")
    .schema(company_industries_schema)
    .csv(company_industries_path)
)

employee_counts_df = (
    spark.read.option("header", "true")
    .option("multiLine", "true")
    .schema(employee_counts_schema)
    .csv(employee_counts_path)
)


# 4. 중복 제거: groupby(company_id) -> agg -> max(time_recorded)
agg_latest_employee_counts_df = employee_counts_df.groupBy("company_id").agg(
    max("time_recorded").alias("time_recorded")
)


# 5. employee_count 컬럼 추가
latest_employee_counts_df = agg_latest_employee_counts_df.join(
    employee_counts_df,
    on=["company_id", "time_recorded"],
    how="inner",
)


# 6. 필터링: 업종이 IT Services and IT Consulting인 회사 추출
industry = "IT Services and IT Consulting"
it_services_company_industries_df = company_industries_df.filter(
    col("industry") == industry
)

# 7.조인: 회사 정보와 종업원 수 결합 및 필터링: 종업원수 1000명 이상 및 정렬: 종업원수 내림차순 정렬
result_df = (
    it_services_company_industries_df.join(
        latest_employee_counts_df,
        on="company_id",
        how="inner",
    )
    .filter(col("employee_count") >= 1000)
    .orderBy(col("employee_count").desc())
)

# 8. 'company_id', 'employee_count', 컬럼만 출력
result_df.select("company_id", "employee_count").show()

time.sleep(300)

순서도

이미지가 압축되어 올라가 [링크] 첨부합니다.