Я хочу создать по DataFrame
указанной схеме в Scala. Я пробовал использовать чтение JSON (я имею в виду чтение пустого файла), но не думаю, что это лучшая практика.
scala
apache-spark
dataframe
apache-spark-sql
user1735076
источник
источник
Начиная с Spark 2.0.0, вы можете делать следующее.
Класс дела
Определим
Person
класс case:scala> case class Person(id: Int, name: String) defined class Person
spark
Неявный импорт SparkSessionEncoders
: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» (см. Функции поддержки для DataFrames в org.apache.spark.sql.ColumnName ).
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)
источник
spark.emptyDataset
нет в моем модуле, как его использовать? есть ли (правильные) похожие на (неправильные)val df = apache.spark.emptyDataset[RawData]
?spark
- это ценность, которую вы создаете,SparkSession.builder
не являясь частьюorg.apache.spark
пакета.spark
Используются два имени. Это то, что уspark
вас есть прямоspark-shell
из коробки.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 и передать пустой 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") }
источник
Версия Java для создания пустого DataSet:
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; }
источник
Вот решение, которое создает пустой фрейм данных в 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)
источник
Это полезно для тестирования.
Seq.empty[String].toDF()
источник
Начиная с Spark 2.4.3
val df = SparkSession.builder().getOrCreate().emptyDataFrame
источник