조금 다름
텍스트를 로딩해서 regex를 이용해 파싱.
구조화된 데이터 프레임으로 바꿔본다.

컬러 코딩한 부분을 빼고 나머지는 항상 같은 문자열. 추출을 해본다.
regex는
\S+
\d 숫자
언어마다 조금씩 다르지만 비슷함. regex를 통해 컬러코딩된 부분을 추출해서, 스트럭쳐 데이터로 추출해보고 싶은 것임.

conf 오브젝트로 간단한 환경설정
이어서 스파크세션 객체 생성

txt를 받아봄
ls -tl로 확인
head -5 transfer_cost.txt로

추출 대상 확인.
먼저 로딩을 해야함

저 자체를 스트링, 파일을 로딩함.
이제는 text를 써볼 예정

schema = SturctType([ StructField("text", StringType(), True)])
transfer_cost_df = spark.read.schema(schema).text("transfer_cost.txt")

transfer_cost_df.show(truncate=False)

truncate False는 필드 다 보여달라는 것임. 자르지 말고

여기에 regex를 적용해본다.
인풋 데이터프레임이 로딩됨.
여기서 추출해서 새로운 데이터프레임.
데이터프레임 지원 함수중 withCoulmun으로 새로운 컬럼, 존재하는 컬럼을 바꿀 수 있음. 그게 첫번째 파라미터. 새로운 컬럼 이름, 또는 존재하는 컬럼 이름.
두번째 파라미터, 컬럼의 값. 어떤 값으로 채울 것인가 보통 여기 UDF(유저 정의함수)가 들어감. 우리는 pyspark의 함수를 가져다 쓸것임. regexp_extract함수를 쓸 것임. 파라미터는 대상이 되는 컬럼의 이름, 두번째는 어떤 형태의 추출을 할것인지 regex_str을, 세번째는 몇번째 매칭되는지 주의. 1부터 시작임.
이걸 5번 반복해서
5번 빌드업 패턴을 반복해서 새로운 컬럼을 만드는 것임.

텍스트 하나만 있었는데, 5개의 새로운 필드가 추가된 것을 확인할수 있고, text를 불필요 함수를 없앤다.
drop으로 없앰.

최종적으로 원하는 파일이 만들어졌는데 csv로 만들어 봄.

그냥 로컬에 파일로 저장
.wirte.csv("~.csv")

잘 보면 파일이 아니고 디렉토리임.
보통 데이터프레임이 굉장히 클 수 있음. HDFS의 저장시 데이터 블록 128MB로 저장. 디렉토리 단위로 만들어지고 들어가면 해당 데이터프레임내용이 여러개의 파일로 저장될 수 있음.
이번엔 파일 하나로 들어가게 됨.

part로 시작함. 첫번째 데이터 블록임.
맵리듀스, 워드카운트 예제시 만들어진 적이 있었음.

head -5 패쓰.csv 로 실행시켜보면

5개의 레코드가 정규표현식에 맞춰 추출된 것을 확인.
이번엔 csv가 아닌 json으로 저장.

긴 형태로 저장
.write.format("json").save("~~~.json") 또는 위치

part패스를 복사해다가 head로 출력해보면

정규표현식

수업 때 정규 표현식은 가볍게 넘어 갔는데 일단 캡쳐해놓음.

from pyspark.sql.functions import *
regex_str = r'On (\S+) the cost per ton from (\d+) to (\d+) is (\S+) at (.*)'

df_with_new_columns = transfer_cost_df\
    .withColumn('week', regexp_extract('text', regex_str, 1))\
    .withColumn('departure_zipcode', regexp_extract(column('text'), regex_str, 2))\
    .withColumn('arrival_zipcode', regexp_extract(transfer_cost_df.text, regex_str, 3))\
    .withColumn('cost', regexp_extract(col('text'), regex_str, 4))\
    .withColumn('vendor', regexp_extract(col('text'), regex_str, 5))

나중에 스트링 단위에서 추출할 때 참고하자.