스파크 - CSV 파일을 DataFrame으로 로드할 수 있나요?

java.lang.RuntimeException: hdfs:///csv/file/dir/file.csv은 Parquet 파일이 아닙니다. 꼬리 부분에 기대되는 매직 넘버 [80, 65, 82, 49] 대신 [49, 59, 54, 10]이 발견되었습니다.
    at parquet.hadoop.ParquetFileReader.readFooter(ParquetFileReader.java:418)
    at org.apache.spark.sql.parquet.ParquetRelation2$MetadataCache$$anonfun$refresh$6.apply(newParquet.scala:277)
    at org.apache.spark.sql.parquet.ParquetRelation2$MetadataCache$$anonfun$refresh$6.apply(newParquet.scala:276)
    at scala.collection.parallel.mutable.ParArray$Map.leaf(ParArray.scala:658)
    at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply$mcV$sp(Tasks.scala:54)
    at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply(Tasks.scala:53)
    at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply(Tasks.scala:53)
    at scala.collection.parallel.Task$class.tryLeaf(Tasks.scala:56)
    at scala.collection.parallel.mutable.ParArray$Map.tryLeaf(ParArray.scala:650)
    at scala.collection.parallel.AdaptiveWorkStealingTasks$WrappedTask$class.compute(Tasks.scala:165)
    at scala.collection.parallel.AdaptiveWorkStealingForkJoinTasks$WrappedTask.compute(Tasks.scala:514)
    at scala.concurrent.forkjoin.RecursiveAction.exec(RecursiveAction.java:160)
    at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

Apache Spark에서 CSV 파일을 DataFrame으로 로드하는 올바른 명령어는 무엇인가요?

답변 1

spark-csv는 핵심 Spark 기능의 일부이며 별도의 라이브러리가 필요하지 않습니다.
따라서 다음과 같이 진행할 수 있습니다.

df = spark.read.format(csv).option(header, true).load(csvfile.csv)

스칼라에서는(이는 csv에 대한 ,를 구분자로 사용하는 등 모든 형식에 적용되는 방법입니다)

val df = sqlContext.read.format(com.databricks.spark.csv)
.option(delimiter, ,)
.load(csvfile.csv)

답변 2

스파크(Spark) - CSV 파일을 DataFrame으로 로드하기
스파크(Spark)는 대규모 데이터 처리와 분석을 위한 오픈 소스 클러스터 컴퓨팅 프레임워크로, 빠른 처리 속도와 확장성을 제공합니다. 스파크는 다양한 데이터 소스와 연동할 수 있는데, CSV 파일도 그 중 하나입니다. 이번 글에서는 CSV 파일을 스파크의 DataFrame으로 로드하는 방법에 대해 알아보겠습니다.
CSV(Comma-Separated Values: 콤마로 구분된 값) 파일은 텍스트 파일 형식의 데이터 저장 방식으로 가장 일반적으로 사용되는 파일 형식 중 하나입니다. 간단한 구조와 편리한 데이터 공유로 인해 많은 데이터 과학 프로젝트에서 사용되고 있습니다. 이러한 이유로 CSV 파일을 스파크로 로드하여 데이터 처리 및 분석을 수행하는 것은 매우 유용합니다.
스파크는 다양한 데이터 형식을 처리할 수 있는데, CSV 파일을 로드하기 위해서는 `spark.read.csv()` 메서드를 사용합니다. 아래는 스파크를 사용하여 CSV 파일을 DataFrame으로 로드하는 예제 코드입니다.
```python
from pyspark.sql import SparkSession
# 스파크 세션 생성
spark = SparkSession.builder \
.appName(CSV to DataFrame) \
.getOrCreate()
# CSV 파일 로드
df = spark.read.csv(파일경로.csv, header=True, inferSchema=True)
# DataFrame 출력
df.show()
```
위의 코드에서 `spark.read.csv()` 메서드를 사용하여 로드할 CSV 파일의 경로를 지정하고, `header=True`와 `inferSchema=True`를 설정하여 첫 번째 행을 헤더로 사용하고 열의 데이터 유형을 추론합니다. 마지막으로 `df.show()`를 사용하여 로드한 DataFrame의 내용을 출력합니다.
스파크를 사용하여 CSV 파일을 로드하면 데이터 처리 및 분석에 필요한 다양한 기능과 함수를 활용할 수 있습니다. 예를 들어, `groupBy()`, `agg()`, `filter()`와 같은 함수를 사용하여 데이터를 집계, 필터링할 수 있으며, SQL 쿼리와 유사한 방식으로 데이터에 접근할 수도 있습니다.
이렇듯, 스파크를 사용하여 CSV 파일을 DataFrame으로 로드하는 방법에 대해 알아보았습니다. CSV 파일은 일반적으로 사용되는 데이터 형식 중 하나이며, 스파크를 사용하여 이를 처리할 수 있다는 것은 데이터 과학 및 빅데이터 분석에 큰 도움이 될 것입니다.