Streaming에서 Kafka 데이타 추출하기 부분 진행하고 있는데요 spark_ kafka.py 실행시 에러가 납니다. ㅠ root@81599cbd6b8f:/opt/bitnami/spark/work# spark-submit --master spark://spark:7077 --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.4.1 spark_kafka.py ..... --------------------------------------------------------------------- | | modules || artifacts | | conf | number| search|dwnlded|evicted|| number|dwnlded| --------------------------------------------------------------------- | default | 11 | 11 | 11 | 0 || 11 | 11 | --------------------------------------------------------------------- :: retrieving :: org.apache.spark#spark-submit-parent-8f3a0b4c-b23d-4dfa-b9b0-8649735433fc confs: [default] 11 artifacts copied, 0 already retrieved (56445kB/64ms) 23/11/18 23:57:21 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable 23/11/18 23:57:24 WARN ResolveWriteToStream: spark.sql.adaptive.enabled is not supported in streaming DataFrames/Datasets and will be disabled. 23/11/18 23:57:24 WARN OffsetSeqMetadata: Updating the value of conf 'spark.sql.shuffle.partitions' in current session from '3' to '200'. 23/11/18 23:57:24 ERROR MicroBatchExecution: Query [id = 40288f62-daae-4e69-80db-ff6f83156268, runId = 535853f9-9153-44be-8eca-19f75ee8b4ea] terminated with error java.lang.IllegalArgumentException: Expected e.g. {"topicA":{"0":23,"1":-1},"topicB":{"0":-2}}, got {"logOffset":2} at org.apache.spark.sql.kafka010.JsonUtils$.partitionOffsets(JsonUtils.scala:75) at org.apache.spark.sql.kafka010.KafkaMicroBatchStream.deserializeOffset(KafkaMicroBatchStream.scala:216) at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$getStartOffset$1(MicroBatchExecution.scala:454) at scala.Option.map(Option.scala:230) at org.apache.spark.sql.execution.streaming.MicroBatchExecution.getStartOffset(MicroBatchExecution.scala:454) at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$4(MicroBatchExecution.scala:489) at org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:411) at org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:409) at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67) at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$2(MicroBatchExecution.scala:488) at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:286) at scala.collection.Iterator.foreach(Iterator.scala:943) at scala.collection.Iterator.foreach$(Iterator.scala:943) at scala.collection.AbstractIterator.foreach(Iterator.scala:1431) at scala.collection.IterableLike.foreach(IterableLike.scala:74) at scala.collection.IterableLike.foreach$(IterableLike.scala:73) at scala.collection.AbstractIterable.foreach(Iterable.scala:56) at scala.collection.TraversableLike.map(TraversableLike.scala:286) at scala.collection.TraversableLike.map$(TraversableLike.scala:279) at scala.collection.AbstractTraversable.map(Traversable.scala:108) at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$1(MicroBatchExecution.scala:477) at scala.runtime.java8.JFunction0$mcZ$sp.apply(JFunction0$mcZ$sp.java:23) at org.apache.spark.sql.execution.streaming.MicroBatchExecution.withProgressLocked(MicroBatchExecution.scala:802) at org.apache.spark.sql.execution.streaming.MicroBatchExecution.constructNextBatch(MicroBatchExecution.scala:473) at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$2(MicroBatchExecution.scala:266) at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23) at org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:411) at org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:409) at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67) at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$1(MicroBatchExecution.scala:247) at org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:67) at org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:237) at org.apache.spark.sql.execution.streaming.StreamExecution.$anonfun$runStream$1(StreamExecution.scala:306) at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23) at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:827) at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:284) at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:207) Traceback (most recent call last): File "/opt/bitnami/spark/work/spark_kafka.py", line 38, in <module> query.awaitTermination() File "/opt/bitnami/spark/python/lib/pyspark.zip/pyspark/sql/streaming/query.py", line 201, in awaitTermination File "/opt/bitnami/spark/python/lib/py4j-0.10.9.7-src.zip/py4j/java_gateway.py", line 1322, in __call__ File "/opt/bitnami/spark/python/lib/pyspark.zip/pyspark/errors/exceptions/captured.py", line 175, in deco pyspark.errors.exceptions.captured.StreamingQueryException: [STREAM_FAILED] Query [id = 40288f62-daae-4e69-80db-ff6f83156268, runId = 535853f9-9153-44be-8eca-19f75ee8b4ea] terminated with exception: Expected e.g. {"topicA":{"0":23,"1":-1},"topicB":{"0":-2}}, got {"logOffset":2}
판다스 Index객체 이해 강의에서 사용한 예제를 그대로 적용하니 문제가 생겨 질문드립니다 reset_index()에 업데이트가 진행되어 칼럼명을 자동으로 바꿔주는 것 같은데 맞나요? value_counts()를 통해 titanic_df['Pclass']의 객체를 얻으면 기존에는 Name:Pclass 로 출력되었던 것 같은데, 이제는 value_counts()를 통해 얻은 모든 객체의 Name이 count로 표기되고 이렇게 name이 count인 객체에 한해 reset_index()를 적용했을 때 칼럼명을 자동으로 적절히 바꾸어주는 것 같습니다
강의를 듣다보니 범주형 데이터일 때, Label Encoding을 하는 방법이 2가지가 있더라고요 from sklearn.preprocessing import LabelEncoder astype('category'). cat.codes 2가지 경우가 결과물에 어떻게 영향을 미치게 될까요? 일단 한 문제로 두 방법으로 인한 roc_auc_score를 검증했을 땐, 유의차가 없었습니다.
실제 캐글(Kaggle) 문제 풀고, 제출해보며, 성능 개선 기법 익히기5 강의에서 질문입니다. 해당 강의 2:50 부분부터 Evaluation 코드에 관해 설명을 해주셨는데요, with torch.no _grad() 안쪽의 코드에서는 with torch.no_grad(): model.eval() ret = None for img, fileid in data_loader: img = img.to(device) pred = model(img) ids += list(fileid) # ex) [1, 2] + [3, 4] -> [1, 2, 3, 4] if ret is None: ret = pred.cpu().numpy() else: ret = np.vstack([ret, pred.cpu().numpy()]) return ret, ids pred.cpu().numpy()와 같이 detach()를 따로 사용하지 않은 걸 볼 수 있었습니다. detach()의 사용에 관해서 찾아보니, with torch.no _grad() 블록은 역전파(gradient 계산)를 비활성화 하므로, pred는 gradient 계산과 관련이 없기에, detach()를 따로 호출하지 않고 numpy 배열로 변환할 수 있다 라는 내용을 찾을 수 있었습니다. 그 전에 수업 시간에 보여주신 코드에서는 model.eval() with torch.no_grad(): for x_minibatch in x_test_batch_list: y_minibatch_pred = model(x_minibatch) print(len(y_minibatch_pred), y_minibatch_pred.shape) y_pred_list.extend(y_minibatch_pred.squeeze().detach().tolist()) 이와 같이 with torch.no _grad() 안에서도 detach()를 사용하신 코드들을 볼 수 있었는데요, Evaluation 단계에서 detach()의 사용이 필수적인 것인지 여쭙고 싶습니다!
보통 작업형2에서는 예측값을 물어보는 문제가 나오던데요. 문제가 나올때 어느문제는 분류모델을, 어느문제는 회귀모델을 사용해야하는지 궁금합니다. 지금까지 강의+기출문제를 보면서는 분류/회귀를 결정하는 부분이 평가 모델을 통해 진행된다는 느낌을 받았는데요. 1) roc_auc_score, accuracy_score 이 평가모델로 쓰일 경우, 분류형 모델 사용(Classifier) 2) rmse, mean_squared_error 이 평가모델로 쓰일 경우, 회귀모델 사용(Regressor) 이렇게 생각하면서 작업형2를 접근하는게 맞는지 궁금합니다.
학습 관련 질문을 남겨주세요. 상세히 작성하면 더 좋아요! 질문과 관련된 영상 위치를 알려주면 더 빠르게 답변할 수 있어요 먼저 유사한 질문이 있었는지 검색해보세요 라벨인코딩 le 선언 관련 질문 라벨 인코딩 시에 le를 for문 밖과 안 모두에서 선언하는 이유가 있을까요? 각 col마다 새로운 le로 학습하고 변환해야하나요? 원핫인코딩 시 아래 동작 여부의 이유가 궁급합니다. pd.get_dummies 함수를 사용할 때 c_train과 c_train[cols]를 넣었을 때 1)은 c_train만 동작하고 2)는 둘 다 동작합니다. 1) 직접 작성 시, cols를 X_train으로 뽑아냈을 때 오류 2) 강의 시, c_train으로 cols 뽑아냈을 때는 동작
선생님 혹시 컬럼에 float값이 있으면 무조건 민맥스 스케일링 사용하나요? 3회는 사용 하셨고 6회 풀이에는 넘어가셨길래 궁금합니다! 그리고 마지막에 제출할때 수험번호 안쓰고 (6회 풀이 입니다) 예를 들어 pd.DataFrame({'index':test.index,'y_pred':pred}).to_csv('0000.csv', index=False)요렇게 했는데 submit.to _csv( 'result.csv' , index = False ) 이렇게 제출하는 형식도 있는건가요?
