지정된 스키마로 빈 DataFrame을 만드는 방법은?

지정된 스키마로 빈 DataFrame을 만드는 방법은?

나는 만들기를 하고 싶습니다.DataFrame스칼라에서 지정된 스키마를 사용합니다.JSON read(빈 파일을 읽는 것을 의미함)를 사용해 보았지만 그게 최선의 방법은 아닌 것 같습니다.

다음 스키마를 사용하는 데이터 프레임을 원하는 경우를 가정합니다.

root
 |-- k: string (nullable = true)
 |-- v: integer (nullable = false)

단순히 데이터 프레임에 대한 스키마를 정의하고 빈 상태로 사용하면 됩니다.RDD[Row]:

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

val schema = StructType(
    StructField("k", StringType, true) ::
    StructField("v", IntegerType, false) :: Nil)

// Spark < 2.0
// sqlContext.createDataFrame(sc.emptyRDD[Row], schema) 
spark.createDataFrame(sc.emptyRDD[Row], schema)

PySpark 등가물은 거의 같습니다.

from pyspark.sql.types import StructType, StructField, IntegerType, StringType

schema = StructType([
    StructField("k", StringType(), True), StructField("v", IntegerType(), False)
])

# or df = sc.parallelize([]).toDF(schema)

# Spark < 2.0 
# sqlContext.createDataFrame([], schema)
df = spark.createDataFrame([], schema)

암시적 인코더 사용(Scala 전용)Product과 같은 유형Tuple:

import spark.implicits._

Seq.empty[(String, Int)].toDF("k", "v")

또는 케이스 클래스:

case class KV(k: String, v: Int)

Seq.empty[KV].toDF

아니면

spark.emptyDataset[KV].toDF

Spark 2.0.0 시점에서 다음과 같은 작업을 수행할 수 있습니다.

케이스 클래스

A를 정의해 보겠습니다.Person케이스 클래스:

scala> case class Person(id: Int, name: String)
defined class Person

수입품spark스파크 세션 암시적Encoders:

scala> import spark.implicits._
import spark.implicits._

그리고 SparkSession을 사용하여 빈 공간을 만듭니다.Dataset[Person]:

scala> spark.emptyDataset[Person]
res0: org.apache.spark.sql.Dataset[Person] = [id: int, name: string]

스키마 DSL

스키마 "DSL"을 사용할 수도 있습니다(org.apache.spark.sql의 DataFrame 지원 기능 참조).열 이름).

scala> val id = $"id".int
id: org.apache.spark.sql.types.StructField = StructField(id,IntegerType,true)

scala> val name = $"name".string
name: org.apache.spark.sql.types.StructField = StructField(name,StringType,true)

scala> import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructType

scala> val mySchema = StructType(id :: name :: Nil)
mySchema: org.apache.spark.sql.types.StructType = StructType(StructField(id,IntegerType,true), StructField(name,StringType,true))

scala> import org.apache.spark.sql.Row
import org.apache.spark.sql.Row

scala> val emptyDF = spark.createDataFrame(sc.emptyRDD[Row], mySchema)
emptyDF: org.apache.spark.sql.DataFrame = [id: int, name: string]

scala> emptyDF.printSchema
root
 |-- id: integer (nullable = true)
 |-- name: string (nullable = true)

빈 DataSet을 만드는 Java 버전:

public Dataset<Row> emptyDataSet(){

    SparkSession spark = SparkSession.builder().appName("Simple Application")
                .config("spark.master", "local").getOrCreate();

    Dataset<Row> emptyDataSet = spark.createDataFrame(new ArrayList<>(), getSchema());

    return emptyDataSet;
}

public StructType getSchema() {

    String schemaString = "column1 column2 column3 column4 column5";

    List<StructField> fields = new ArrayList<>();

    StructField indexField = DataTypes.createStructField("column0", DataTypes.LongType, true);
    fields.add(indexField);

    for (String fieldName : schemaString.split(" ")) {
        StructField field = DataTypes.createStructField(fieldName, DataTypes.StringType, true);
        fields.add(field);
    }

    StructType schema = DataTypes.createStructType(fields);

    return schema;
}

import scala.reflect.runtime.{universe => ru}
def createEmptyDataFrame[T: ru.TypeTag] =
    hiveContext.createDataFrame(sc.emptyRDD[Row],
      ScalaReflection.schemaFor(ru.typeTag[T].tpe).dataType.asInstanceOf[StructType]
    )
  case class RawData(id: String, firstname: String, lastname: String, age: Int)
  val sourceDF = createEmptyDataFrame[RawData]

여기서는 StructType을 사용하여 scala에서 스키마를 생성하고 Empty RDD를 전달하여 빈 테이블을 생성할 수 있습니다.아래 코드는 동일한 코드입니다.

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql._
import org.apache.spark.sql.Row
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.IntegerType
import org.apache.spark.sql.types.BooleanType
import org.apache.spark.sql.types.LongType
import org.apache.spark.sql.types.StringType



//import org.apache.hadoop.hive.serde2.objectinspector.StructField

object EmptyTable extends App {
  val conf = new SparkConf;
  val sc = new SparkContext(conf)
  //create sparksession object
  val sparkSession = SparkSession.builder().enableHiveSupport().getOrCreate()

  //Created schema for three columns 
   val schema = StructType(
    StructField("Emp_ID", LongType, true) ::
      StructField("Emp_Name", StringType, false) ::
      StructField("Emp_Salary", LongType, false) :: Nil)

      //Created Empty RDD 

  var dataRDD = sc.emptyRDD[Row]

  //pass rdd and schema to create dataframe
  val newDFSchema = sparkSession.createDataFrame(dataRDD, schema)

  newDFSchema.createOrReplaceTempView("tempSchema")

  sparkSession.sql("create table Finaltable AS select * from tempSchema")

}

이것은 테스트 목적에 도움이 됩니다.

Seq.empty[String].toDF()

pyspark 2.0.0 이상에서 빈 데이터 프레임을 생성하는 솔루션이 있습니다.

from pyspark.sql import SQLContext
sc = spark.sparkContext
schema = StructType([StructField('col1', StringType(),False),StructField('col2', IntegerType(), True)])
sqlContext.createDataFrame(sc.emptyRDD(), schema)

나는 이미 데이터 프레임을 가지고 있다는 특별한 요구사항이 있었지만 일정한 조건을 감안하여 빈 데이터 프레임을 반환해야 했기 때문에 반환했습니다.df.limit(0)대신.

아직 언급되지 않은 다음 구문을 추가하고자 합니다.

Seq[(String, Integer)]().toDF("k", "v")

그것은 그들이 다음과 같이()part는 값을 위한 것입니다.데이터 프레임이 비어 있어서 데이터 프레임이 비어 있습니다.

이 구문은 다음을 추가하는 데도 유용합니다.null값을 수동으로 지정합니다.다른 옵션들은 그렇지 않거나 지나치게 장황합니다.

스파크 2.4.3 기준

val df = SparkSession.builder().getOrCreate().emptyDataFrame

언급URL : https://stackoverflow.com/questions/31477598/how-to-create-an-empty-dataframe-with-a-specified-schema