Colab에 pyspark가 미리 설치되어 있지는 않지만!pip install pyspark명령으로 설치 후 사용 가능하더라구요.Colab에 자바와 하둡, 스파크가 이미 설치되어 있는 것으로 생각해도 될까요? 제가 강의로 이해하기로는하둡에 있는 일부 기능- 맵리듀스를 스파크가 대체할 수 있는데,기존 하둡과 호환성을 위해 다른 부분은 하둡 그대로 두고 그 부분만 스파크로 대체해야 하고그래서 하둡도 필요하고 스파크는 자바로 되어 있으니 자바도 필요하고,스파크 설치 후 이것을 파이썬으로 래핑한 pyspark를 실행하는 것으로그렇게 생각하고 있는데 이게 맞는지 모르겠습니다. SparkContext가 생성되면서 내부적으로 스파크 어플리케이션이 실행되나요? 아니면 데몬같은 것이 실행중인 상태인 걸까요? 너무 모른 채로 질문드려서 죄송합니다.

안녕하세요. 강사님, 수업을 잘 듣고 있습니다! 다름이 아니라 spark_xgboost_lightgbm에서 버전이 바뀌면서앞선 댓글을 통해 SparkXGBClassifier로 수정하여해결은 하였으나, 새로 바뀐 XGB에서는xgb_model.feature_importances 이부분이 나오질 않아서, 혹시 SparkXGBClassifier의 featureImportances를 보려면 어떻게 해야하는지 from mmlspark.lightgbm import LightGBMClassifierimport mlflowlgbm_classifier = LightGBMClassifier(featuresCol="features", labelCol="label", numLeaves=100, maxDepth=10, numIterations=100)lgbm_model = lgbm_classifier.fit(train_sdf)-> 이부분부터 값이 나오질 않습니다.오류메시지org.apache.spark.SparkException: Job aborted due to stage failure: Task 4 in stage 73.0 failed 1 times, most recent failure: Lost task 4.0 in stage 73.0 (TID 285) (ip-번호.us-west-2.compute.internal executor driver): java.lang.NoSuchMethodError: com.microsoft.ml.lightgbm.lightgbmlib.new_doubleArray(J)Lcom/microsoft/ml/lightgbm/SWIGTYPE_p_double;이 두부분에 대해서 문의 드립니다! 

안녕하십니까 수업 잘 수강하고 있습니다!수업을 수강하며 pyspark를 통해 예전에 했던 프로젝트의 데이터를 전처리부터 머신러닝까지 적용해보는 중인데, 전처리 과정에서 데이터를 수정해야 하는 경우에 대하여 질문이 있습니다.spark의 경우 pandas처럼 바꾸고 싶은 컬럼의 값만 바꿀 수 있지가 않고, withColumn을 통해 새로운 컬럼을 만들어내는 형식으로 대체가 가능한걸로 알고 있습니다.단순히 컬럼별로 기준을 정해서 바꾸는 거면 withColumn으로도 가능하지만, 만약 개별 줄마다 값을 변경해야 할 경우 for문을 써서 바꿔야 하는 경우가 있는데, 이렇게 할 경우 제 현재 작업 환경(로컬)이 노드가 하나여서 그런지 Java.lang.OutOfMemoryError이 뜨더군요. 그래서 기존에 하던 pandas에서 하던 것 처럼 하려면, toPandas로 바꿔서 해도 되긴 합니다만.. 그러면 pyspark를 이 단계에서는 굳이 사용해봐야 의미가 없고, 또한 나중에 in-memory에서는 처리를 하기 힘든 큰 데이터의 경우에서는 pandas를 사용하지 못하니 방법이 아예 사라지게 됩니다. 그래서 질문은, 만약 이 경우처럼 세세하게 한줄한줄마다 값을 수정할 필요가 있을 경우, 어떻게 하는 것이 좋을지 궁금합니다. 또, spark dataframe을 toPandas로 변환할 경우 그냥 Pandas dataframe으로 불러오는 것과 차이점이 있는지도 여쭤보고 싶습니다.

안녕하세요! 좋은 강의 잘 듣고 있습니다.제공해주신 연습 예제부터 차근차근 실습해보고 더 큰 데이터로 학습 해보고자 컬럼수 10개 / 큰 용량의 csv파일로 학습중입니다.  변수만 바꾸고 코드를 동일하고 학습하고 있습니다.코드의 마지막 줄인results = avg_by_count.collect() print(results)이 부분에서 에러가 나서 어떻게 해결해야 하는지 질문드립니다. === 이것저것 검색해보다가 pyspark / python 버전이 안맞으면 에러가 날 수 있다고 보고 버전도 확인해봤습니다.print(sc.version)# 3.3.2print(sc.pythonVer)#3 .10print(sc.master)#local[*]   아래는 에러 전문입니다.1197 sock_info = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd()) 1198 return list(_load_from_socket(sock_info, self._jrdd_deserializer)) File /usr/local/spark/python/lib/py4j-0.10.9.5-src.zip/py4j/java_gateway.py:1321, in JavaMember.__call__(self, *args) 1315 command = proto.CALL_COMMAND_NAME +\ 1316 self.command_header +\ 1317 args_command +\ 1318 proto.END_COMMAND_PART 1320 answer = self.gateway_client.send_command(command) -> 1321 return_value = get_return_value( 1322 answer, self.gateway_client, self.target_id, self.name) 1324 for temp_arg in temp_args: 1325 temp_arg._detach() File /usr/local/spark/python/lib/py4j-0.10.9.5-src.zip/py4j/protocol.py:326, in get_return_value(answer, gateway_client, target_id, name) 324 value = OUTPUT_CONVERTER[type](answer[2:], gateway_client) 325 if answer[1] == REFERENCE_TYPE: --> 326 raise Py4JJavaError( 327 "An error occurred while calling {0}{1}{2}.\n". 328 format(target_id, ".", name), value) 329 else: 330 raise Py4JError( 331 "An error occurred while calling {0}{1}{2}. Trace:\n{3}\n". 332 format(target_id, ".", name, value)) Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe. : org.apache.spark.SparkException: Job aborted due to stage failure: Task 1 in stage 10.0 failed 1 times, most recent failure: Lost task 1.0 in stage 10.0 (TID 11) (b4b9f5895184 executor driver): org.apache.spark.api.python.PythonException: Traceback (most recent call last): File "/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py", line 686, in main process() File "/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py", line 676, in process out_iter = func(split_index, iterator) File "/usr/local/spark/python/pyspark/rdd.py", line 3472, in pipeline_func return func(split, prev_func(split, iterator)) File "/usr/local/spark/python/pyspark/rdd.py", line 3472, in pipeline_func return func(split, prev_func(split, iterator)) File "/usr/local/spark/python/pyspark/rdd.py", line 540, in func return f(iterator) File "/usr/local/spark/python/pyspark/rdd.py", line 2554, in combineLocally merger.mergeValues(iterator) File "/usr/local/spark/python/lib/pyspark.zip/pyspark/shuffle.py", line 253, in mergeValues for k, v in iterator: File "/usr/local/spark/python/lib/pyspark.zip/pyspark/util.py", line 81, in wrapper return f(*args, **kwargs) File "/tmp/ipykernel_35939/1438163465.py", line 11, in parse_line ValueError: invalid literal for int() with base 10: '61.760999927297242' at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:552) at org.apache.spark.api.python.PythonRunner$$anon$3.read(PythonRunner.scala:758) at org.apache.spark.api.python.PythonRunner$$anon$3.read(PythonRunner.scala:740) at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:505) at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37) at scala.collection.Iterator$GroupedIterator.fill(Iterator.scala:1211) at scala.collection.Iterator$GroupedIterator.hasNext(Iterator.scala:1217) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:140) at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52) at org.apache.spark.scheduler.Task.run(Task.scala:136) at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1504) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) at java.base/java.lang.Thread.run(Thread.java:833) Driver stacktrace: at org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2672) at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2608) at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2607) at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62) at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55) at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49) at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2607) at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:1182) at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:1182) at scala.Option.foreach(Option.scala:407) at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1182) at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2860) at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2802) at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2791) at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49) at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:952) at org.apache.spark.SparkContext.runJob(SparkContext.scala:2238) at org.apache.spark.SparkContext.runJob(SparkContext.scala:2259) at org.apache.spark.SparkContext.runJob(SparkContext.scala:2278) at org.apache.spark.SparkContext.runJob(SparkContext.scala:2303) at org.apache.spark.rdd.RDD.$anonfun$collect$1(RDD.scala:1021) at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112) at org.apache.spark.rdd.RDD.withScope(RDD.scala:406) at org.apache.spark.rdd.RDD.collect(RDD.scala:1020) at org.apache.spark.api.python.PythonRDD$.collectAndServe(PythonRDD.scala:180) at org.apache.spark.api.python.PythonRDD.collectAndServe(PythonRDD.scala) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77) at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.base/java.lang.reflect.Method.invoke(Method.java:568) at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357) at py4j.Gateway.invoke(Gateway.java:282) at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) at py4j.commands.CallCommand.execute(CallCommand.java:79) at py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:182) at py4j.ClientServerConnection.run(ClientServerConnection.java:106) at java.base/java.lang.Thread.run(Thread.java:833) Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last): File "/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py", line 686, in main process() File "/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py", line 676, in process out_iter = func(split_index, iterator) File "/usr/local/spark/python/pyspark/rdd.py", line 3472, in pipeline_func return func(split, prev_func(split, iterator)) File "/usr/local/spark/python/pyspark/rdd.py", line 3472, in pipeline_func return func(split, prev_func(split, iterator)) File "/usr/local/spark/python/pyspark/rdd.py", line 540, in func return f(iterator) File "/usr/local/spark/python/pyspark/rdd.py", line 2554, in combineLocally merger.mergeValues(iterator) File "/usr/local/spark/python/lib/pyspark.zip/pyspark/shuffle.py", line 253, in mergeValues for k, v in iterator: File "/usr/local/spark/python/lib/pyspark.zip/pyspark/util.py", line 81, in wrapper return f(*args, **kwargs) File "/tmp/ipykernel_35939/1438163465.py", line 11, in parse_line ValueError: invalid literal for int() with base 10: '61.760999927297242' at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:552) at org.apache.spark.api.python.PythonRunner$$anon$3.read(PythonRunner.scala:758) at org.apache.spark.api.python.PythonRunner$$anon$3.read(PythonRunner.scala:740) at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:505) at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37) at scala.collection.Iterator$GroupedIterator.fill(Iterator.scala:1211) at scala.collection.Iterator$GroupedIterator.hasNext(Iterator.scala:1217) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:140) at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52) at org.apache.spark.scheduler.Task.run(Task.scala:136) at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1504) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ... 1 more ㅠㅠ혹시 해결방법을 아신다면 답변 부탁드립니다..감사합니다.!!!!

안녕하세요~ 대학교 한 학기처럼 탄탄하게 빅데이터 파이프라인 관련해서 차근차근 다뤄주시니 도움이 되고 있습니다. Spark Advanced 과정에서 Spark를 자세히 다루면서 학습하고 있습니다 혹시 과제가 있던데? 데이터를 Data Storage에서 배운 HDFS 위에 적재하고 Spark를 돌리면 너무 좋을 것 같은데 혹시 이 부분에 대해서도 가이드 해주시나요?감사합니다.

spark streaming을 설명하시면서 광고로그를 join? 하는걸 예시로 들어주셨는데요, 노출이랑 클릭이랑 interval이 10분~30분 정도가 아니라 7일 이렇게 되는 경우에도 spark streaming으로 처리가 가능한가요?? 궁금합니다.

안녕하세요!강의 수강하면서 HDFS 명령어를 구글링하면서 만져보고 있는데요, 강의 실습에서 hdfs-site.xml에서 replication을 1로 설정해서, hdfs setrep 명령어를 사용해 replication configuration 3으로 변경을 시도했다가 잘 되지 않아서 질문드립니다.w 옵션 없이 setrep 3 / 명령어 입력 시에,./bin/hdfs dfs -stat "%b %F %g %n %o %r %u %y %Y" {파일 경로}결과 %r = 3으로 변경이 되는데./bin/hdfs fsck / -files -blocks결과를 확인해보면, Default replication factor가 여전히 1로 나옵니다.Default replication factor를 변경할 수 있는 방법 질문드립니다. 또 기다려도 hdfs에 올린 파일들이 자동으로 복제가 되지 않고 개수가 계속 1이라(fsck 명령어로 상태 확인 시에 missing replicas 확인됨), 강제로 setrep에 w옵션을 사용해서 기다려봤는데, 계속 복제본을 만들지 않는 결과를 확인했습니다.이게 Pseudo Distributed Operation 모드에서 동작을 시켜서 발생시키는 결과인지, 아니면 어떤 부분이 문제일지 질문드립니다.감사합니다.

안녕하세요, 선생님. 이 글의 제목에 관련된 질문을 드리고 싶습니다.저는 코딩을 할 때 주로 Windows 환경에서 해왔고 그래선지 Python code를 실행시킬 때 script 실행을 하는 경험이 없었고 함수 지향 프로그래밍을 하면서 네임 내장 함수를 써가며 script 실행 시 argument를 사용하게 끔 코딩해본 적이 없습니다.본 강의에서 앞선 수업 때 이런 내용에 대해 다뤄주신 적이 있다면 모르겠지만, 그게 아니다보니 선생님께서 이 수업에서 내주신 숙제가 조금 overtraining처럼 느껴집니다. 관련 자료를 뒤져가며 공부하는게 도움이 될 수도 있겠지만 사정이 있어서 빠르게 본 강의를 마스터하고 싶은 마음이 큽니다. 선생님께서 답안을 주시면 그것가지고 공부를 하고 싶습니다. 허락해주신다면요.. 감사합니다. 

안녕하세요, 선생님.지금 섹션2 Data Acquisition[Advanced] 파트를 듣고 있는데수업에서 보여주시는 자료들을 수강생들이 구할 수는 없는지 궁금합니다. 수업 중에 선생님께서 자료에 있는 링크를 누르시면서 "여기 링크를 누르셔서 이 페이지를 확인해보시면~" 와 같은 흐름으로 이어지는 부분이 있는데, 이렇다 보니 자료가 더욱 필요하다고 느껴져 여쭤봅니다.그리고 섹션7에 있는 [실습 및 과제 데이터]를 다운받아서 압축 해제를 시도해보니 10%에서 멈추는 현상이 있습니다. 사용하는 압축해제 프로그램은 7zip입니다. 현재 사용하는 Laptop이 기업용이라 라이센스 문제가 생기지 않는 프로그램을 쓰고 있습니다. 프로그램 문제보단 압축파일 문제이지 않을까 싶은데 확인 부탁드립니다.감사합니다.

안녕하세요, 선생님. 강의에서 사용되는 Linux를 혹시 Docker 위에서 돌려도 괜찮을까요? 제가 Docker 사용에 능숙하지는 못해서 GUI환경은 구축하지 못했는데 혹시 CLI로만 수업을 따라간다고 했을 때 문제가 생기는 점이 있을지 궁금합니다.

EMR 환경에서 spark_app = SparkSession.builder.appName(app_name).getOrCreate() 하면 자동으로 분산처리가 되나요?  아니면 conf를 따로해줘야하는지요? 따로 할경우는 어떻게 conf를 잡으면 될까요?

질문은 많으시면 많을수록 좋습니다. 가능한 빠른 답변 드리겠습니다.원활한 답변을 위해, 자세한 질문 사항 부탁드려요 :D

안녕하세요.mssql에서 벌크 백업한 binary 파일을 spark를 이용하여 parquet, csv 등 파일 변환이 가능한지.. 질문 드립니다. 감사합니다.

안녕하세요.SPARK 2 강의는 언제 오픈할 예정일까요?

안녕하세요. 강의 수강 하면서 많은 도움이 되었습니다.실무에서 Spark ML를 통해 학습한 모델을 저장 후 서빙할 때 어떤 방식으로 진행하는지 궁금하여 질문 드립니다.저는 이전에 파이썬 사이킷런을 이용할 때, tensorflow 도커 이미지를 이용하여, rest api로 웹 서비스에서 사용할 수 있도록 진행 했었는데, Spark ML로 학습한 모델도 동일하게 진행하면 되는지 다른 방식으로 진행하는지 궁금합니다. 또한, 강의에서 LightGBM이 Spark 분산처리로 진행할 때, 성능이 좋지 못하게 나온다 라고 보여주셨는데, 실무에서 분류 계열 지도학습을 할때, 랜덤포레스트 또는 XGBoost를 주로 사용 하시는지도 궁금합니다!감사합니다.

안녕하세요 :-) 수업 잘 듣고 있습니다.다름이 아니라 맥os 환경에서 보여지는 파일 형태가 달라서 문의드립니다.맥 os 환경에서는, order_products_prior.csv.zip 파일을 풀고나면 하나의 csv파일만 나옵니다. 강의 동영상에서 처럼 압축을 푼 뒤에 여러가지의 파일과 폴더가 묶여 나오지 않습니다. 따라서 zip 파일 안에서 csv 파일을 다시 압축을 하고 싶지만 정상적으로 되지 않습니다. 제가 혹시나 해서 압축을 푼 뒤에 다시 csv 파일을 압축해 보았지만 결과는 다른 수강자님께서 겪고 있으셨던 화면과 동일했습니다. (zip 파일안에 다양한 파일이 존재한다는 오류)혹시 제가 잘못하고 있던 것이라면 첨언 부탁드립니다 감사합니다.

안녕하세요 spark 공부하면서 취준하고 있는 대학생입니다.좋은 강의 제공해주셔서 감사합니다 :-)다름이 아니라, 코드에 궁금한 점이 생겨서 문의 드립니다.mmlspark lightgbm 설치 후에 아래에 있는 코드를 실행 해야 하나요? Hyperlink로 접속을 해 봤는데 오류가 있는 거 같습니다.스파크 프레임워크로 처음 공부를 해 봐서 제가 틀린 걸 수도 있습니다. 도움 부탁드립니다:-)

안녕하세요. 좋은 강의 감사합니다.TrainValidationSplit 수강 중 질문이 있습니다.TrainValidationSplit를 이용하여 학습과 검증 비율을 trainRatio 파라미터를 이용하여 나누게 되는데, TrainValidationSplit을 사용할 시 randomSplit은 사용할 필요가 없나요? 아니면 사용하는 경우가 있을까요? CrossValidator은 교차 검증 k fold 방식으로 알고 있는데, randomSplit에서 나눈 비율로 교차 검증을 진행하는 걸로 이해하면 될까요?예를 들면 randomSplit 0.8(학습) 0.2(검증) 으로 나누었다면, 해당 비율로 교차검증을 진행하는지 궁금합니다.마지막으로 현재 업무에서 스파크 스트리밍을 사용하고 있고 scala 기반으로 되어 있기 때문에 scala를 이용한 ml 도입을 고려하고 있습니다. 이때 강의에서 진행하는 pyspark과 비교하여 제공하지 않는 기능이라던지 고려해야 할 사항이 있을까요? 감사합니다!

안녕하세요parquet을 read할때, 파티션 개수는 저장할때의 partition으로 고정될까요??parquet을 read 할때 partiton 1개당 size가 너무 커서, partiton 개수를 늘려서 read 하고 싶습니다. 혹시 방법이 있을까용?  감사합니다.

안녕하세요대용량 데이터 join 방법에 대해 질문드립니다.source A,B,C가 있고 A와 B를 union 하고 C를 조인해야 하는 상황입니다. A, B, C 각각은 모두 대용량 데이터입니다.하지만, 이 코드를 실행하면 spark 내부적으로 C를 2번 read해 A와 C를 조인하고, B와 C를 조인하는 DAG이 생성되는 것을 UI에서 확인했습니다. 이에, C를 1번만 read하게 만들기 위해서 cDF.persist(StorageLevel.DISK_ONLY)를 중간에 삽입해, 원래 의도대로 A와 B를 union하고 C를 조인하도록 DAG을 변경하였습니다. 이런 상황에서 persist를 사용하지 않고 해결할 방법이 있을까요?