CSV 구문 분석 및 Spark 2.x를 사용하여 DataFrame / DataSet으로로드

먼저 기본적으로SparkSession 객체 를 초기화 하여 쉘에서 사용할 수 있습니다.spark

val spark = org.apache.spark.sql.SparkSession.builder
        .master("local") # Change it as per your cluster
        .appName("Spark CSV Reader")
        .getOrCreate;

다음 방법 중 하나를 사용하여 CSV를 DataFrame/DataSet

1. 프로그래밍 방식으로 수행

 val df = spark.read
         .format("csv")
         .option("header", "true") //first line in file has headers
         .option("mode", "DROPMALFORMED")
         .load("hdfs:///csv/file/dir/file.csv")

업데이트 : 나중에 링크가 손상 될 경우를 대비하여 여기 에서 모든 옵션 추가

• 경로 : 파일의 위치. Spark와 유사하게 표준 Hadoop globbing 표현식을 사용할 수 있습니다.
• header : true로 설정하면 파일의 첫 번째 행이 열의 이름을 지정하는 데 사용되며 데이터에 포함되지 않습니다. 모든 유형은 문자열로 간주됩니다. 기본값은 false입니다.
• 구분 기호 : 기본적으로 열은 구분 기호를 사용하지만 구분 기호는 모든 문자로 설정할 수 있습니다
• quote : 기본적으로 따옴표 문자는 “이지만 모든 문자로 설정할 수 있습니다. 따옴표 안의 분리 문자는 무시됩니다.
• escape : 기본적으로 이스케이프 문자는이지만 모든 문자로 설정할 수 있습니다. 이스케이프 된 따옴표는 무시됩니다
• parserLib : 기본적으로 CSV 구문 분석에 해당 라이브러리를 사용 하기 위해 ” univocity “로 설정할 수있는 ” commons “입니다 .
• mode : 파싱 모드를 결정합니다. 기본적으로 PERMISSIVE입니다. 가능한 값은 다음과 같습니다.
  • PERMISSIVE : 모든 행을 구문 분석하려고합니다. 누락 된 토큰에 대해 널이 삽입되고 추가 토큰이 무시됩니다.
  • DROPMALFORMED : 예상보다 적은 수 이상의 토큰이있는 행 또는 스키마와 일치하지 않는 토큰을 삭제합니다.
  • FAILFAST : 잘못된 행 문자 집합이 발견되면 RuntimeException과 함께 중단됩니다. 기본값은 ‘UTF-8’이지만 다른 유효한 문자 집합 이름으로 설정할 수 있습니다
• inferSchema : 열 유형을 자동으로 유추합니다. 데이터를 한 번 더 전달해야하며 기본 주석에서는이 문자로 시작하는 줄을 건너 뜁니다. 기본값은 “#”입니다. 이 값을 null로 설정하여 주석을 비활성화하십시오.
• nullValue : null 값을 나타내는 문자열을 지정합니다.이 문자열과 일치하는 모든 필드는 DataFrame에서 null로 설정됩니다.
• dateFormat : 날짜 또는 타임 스탬프를 읽을 때 사용할 날짜 형식을 나타내는 문자열을 지정합니다. 사용자 정의 날짜 형식은 java.text.SimpleDateFormat의 형식을 따릅니다. 이는 DateType 및 TimestampType 모두에 적용됩니다. 기본적으로 null입니다. 이는 java.sql.Timestamp.valueOf () 및 java.sql.Date.valueOf ()로 시간 및 날짜를 ​​구문 분석하려고 함을 의미합니다.

2. 이 SQL 방식으로도 할 수 있습니다

 val df = spark.sql("SELECT * FROM csv.`hdfs:///csv/file/dir/file.csv`")

종속성 :

 "org.apache.spark" % "spark-core_2.11" % 2.0.0,
 "org.apache.spark" % "spark-sql_2.11" % 2.0.0,

스파크 버전 <2.0

val df = sqlContext.read
    .format("com.databricks.spark.csv")
    .option("header", "true")
    .option("mode", "DROPMALFORMED")
    .load("csv/file/path"); 

종속성 :

"org.apache.spark" % "spark-sql_2.10" % 1.6.0,
"com.databricks" % "spark-csv_2.10" % 1.6.0,
"com.univocity" % "univocity-parsers" % LATEST,

답변

import org.apache.spark.sql.types.{StructType,StructField,StringType,IntegerType};
import org.apache.spark.sql.Row;

val csv = sc.textFile("/path/to/file.csv")
val rows = csv.map(line => line.split(",").map(_.trim))
val header = rows.first
val data = rows.filter(_(0) != header(0))
val rdd = data.map(row => Row(row(0),row(1).toInt))

val schema = new StructType()
    .add(StructField("id", StringType, true))
    .add(StructField("val", IntegerType, true))

val df = sqlContext.createDataFrame(rdd, schema)

val conf = new SparkConf().setMaster("local[2]").setAppName("my app")
val sc = new SparkContext(conf)
val sparkSession = SparkSession.builder
  .config(conf = conf)
  .appName("spark session example")
  .getOrCreate()

val path = "/Users/xxx/Downloads/usermsg.csv"
val base_df = sparkSession.read.option("header","true").
  csv(path)

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.11</artifactId>
    <version>2.0.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql_2.10 -->
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.10</artifactId>
    <version>2.0.0</version>
</dependency>

<!-- https://mvnrepository.com/artifact/org.scala-lang/scala-library -->
<dependency>
    <groupId>org.scala-lang</groupId>
    <artifactId>scala-library</artifactId>
    <version>2.11.8</version>
</dependency>
<dependency>
    <groupId>com.databricks</groupId>
    <artifactId>spark-csv_2.10</artifactId>
    <version>1.4.0</version>
</dependency>

SparkConf conf = new SparkConf().setAppName("JavaWordCount").setMaster("local");
// create Spark Context
SparkContext context = new SparkContext(conf);
// create spark Session
SparkSession sparkSession = new SparkSession(context);

Dataset<Row> df = sparkSession.read().format("com.databricks.spark.csv").option("header", true).option("inferSchema", true).load("hdfs://localhost:9000/usr/local/hadoop_data/loan_100.csv");

        //("hdfs://localhost:9000/usr/local/hadoop_data/loan_100.csv");
System.out.println("========== Print Schema ============");
df.printSchema();
System.out.println("========== Print Data ==============");
df.show();
System.out.println("========== Print title ==============");
df.select("title").show();

### Create a Spark Session
spark = SparkSession.builder.master("local").appName("Classify Urls").getOrCreate()

### Note the options that are used. You may have to tweak these in case of error
html_df = spark.read.csv(html_csv_file_path,
                         header=True,
                         multiLine=True,
                         ignoreLeadingWhiteSpace=True,
                         ignoreTrailingWhiteSpace=True,
                         encoding="UTF-8",
                         sep=',',
                         quote='"',
                         escape='"',
                         maxColumns=2,
                         inferSchema=True)

  /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */

val csvdata = spark.read.options(Map(
    "header" -> "true",
    "ignoreLeadingWhiteSpace" -> "true",
    "ignoreTrailingWhiteSpace" -> "true",
    "timestampFormat" -> "yyyy-MM-dd HH:mm:ss.SSSZZZ",
    "inferSchema" -> "true",
    "mode" -> "FAILFAST"))
  .csv("s3a://landsat-pds/scene_list.gz")