Detaillierte Erläuterung der 4 gängigen Datenquellen in Spark SQL

scala> val df = spark.read.load("hdfs://hadoop001:9000/namesAndAges.parquet")
df: org.apache.spark.sql.DataFrame = [Alter: Bigint, Name: Zeichenfolge]
scala> df.select("Name").schreiben.speichern("Namen.Parkett")

scala> val peopleDF = spark.read.format("json").load("hdfs://hadoop001:9000/people.json")
peopleDF: org.apache.spark.sql.DataFrame = [Alter: Bigint, Name: Zeichenfolge]
scala> peopleDF.write.format("parquet").speichern("hdfs://hadoop001:9000/namesAndAges.parquet")
scala>

val sqlDF = spark.sql("SELECT * FROM parquet.`hdfs://hadoop001:9000/namesAndAges.parquet`")
sqlDF.show()

// Encoder für die gängigsten Typen werden automatisch durch den Import von spark.implicits bereitgestellt._
importiere spark.implicits._
val peopleDF = spark.read.json("Beispiele/src/main/Ressourcen/people.json")
// DataFrames können als Parquet-Dateien gespeichert werden, wobei die Schemainformationen erhalten bleiben
LeuteDF.write.parquet("hdfs://hadoop001:9000/leute.parquet")
// Lesen Sie die oben erstellte Parquet-Datei ein
// Parquet-Dateien sind selbstbeschreibend, sodass das Schema erhalten bleibt
// Das Ergebnis des Ladens einer Parquet-Datei ist auch ein DataFrame
val parquetFileDF = spark.read.parquet("hdfs://hadoop001:9000/people.parquet")
// Parquet-Dateien können auch zum Erstellen einer temporären Ansicht verwendet und dann in SQL-Anweisungen verwendet werden
parquetFileDF.createOrReplaceTempView("parquetFile")
val namesDF = spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19")
namesDF.map(attributes => "Name: " + attribute(0)).show()
// +------------+
// | Wert|
// +------------+
// |Name: Justin|
// +------------+

Weg
└── zu
└── Tabelle
├── Geschlecht=männlich
│ ├── …
│ │
│ ├── Land = USA
│ │ └── Daten.parquet
│ ├── Land=CN
│ │ └── Daten.parquet
│ └── …
└── Geschlecht=weiblich
├── ...
│
├── Land = USA
│ └── Daten.parquet
├── Land=CN
│ └── Daten.parquet
└── …

Wurzel
|-- Name: Zeichenfolge (nullwertfähig = wahr)
|-- Alter: lang (nullwertig = wahr)
|-- Geschlecht: Zeichenfolge (nullwertig = wahr)
|-- Land: Zeichenfolge (nullwertfähig = wahr)

spark.sql.sources.partitionColumnTypeInference.enabled

// In diesem Beispiel wird sqlContext aus dem vorherigen Beispiel verwendet.
// Dies wird verwendet, um ein RDD implizit in einen DataFrame zu konvertieren.
importiere spark.implicits._
//Erstellen Sie einen einfachen DataFrame, der in einem Partitionsverzeichnis gespeichert wird
val df1 = sc.makeRDD(1 bis 5).map(i => (i, i * 2)).toDF("single", "double")
df1.write.parquet("hdfs://hadoop001:9000/data/test_table/key=1")
// Einen weiteren DataFrame in einem neuen Partitionsverzeichnis erstellen,
// Hinzufügen einer neuen Spalte und Löschen einer vorhandenen Spalte
val df2 = sc.makeRDD(6 bis 10).map(i => (i, i * 3)).toDF("single", "triple")
df2.write.parquet("hdfs://hadoop001:9000/data/test_table/key=2")
// Lesen Sie die partitionierte Tabelle
val df3 = spark.read.option("mergeSchema", "true").parquet("hdfs://hadoop001:9000/data/test_table")
df3.printSchema()
// Das endgültige Schema besteht aus allen 3 Spalten in den Parquet-Dateien zusammen
// mit der Partitionierungsspalte, die in den Partitionsverzeichnispfaden angezeigt wurde.
//Wurzel
// |-- einzeln: int (nullwertfähig = wahr)
// |-- double: int (nullwertfähig = wahr)
// |-- triple: int (nullwertfähig = wahr)
// |-- Schlüssel: int (nullwertfähig = wahr)

importiere java.io.File
importiere org.apache.spark.sql.Row
importiere org.apache.spark.sql.SparkSession
Fallklasse Record (Schlüssel: Int, Wert: String)
// warehouseLocation verweist auf den Standardspeicherort für verwaltete Datenbanken und Tabellen
val warehouseLocation = neue Datei("spark-warehouse").getAbsolutePath
val spark = SparkSession
.Erbauer()
.appName("Spark Hive-Beispiel")
.config("spark.sql.warehouse.dir", Lagerort)
.enableHiveSupport()
.getOrCreate()
importiere spark.implicits._
spark.sql importieren
sql("TABELLE ERSTELLEN, WENN NICHT VORHANDEN src (Schlüssel INT, Wert STRING)")
sql("DATEN LOKALEN INPATH 'examples/src/main/resources/kv1.txt' IN TABELLE src laden")
// Abfragen werden in HiveQL ausgedrückt
sql("SELECT * FROM src").anzeigen()
// +---+--------+
// |Schlüssel| Wert|
// +---+--------+
//|238|val_238|
//| 86| Wert_86|
//|311|val_311|
// ...
// Aggregationsabfragen werden ebenfalls unterstützt.
sql("SELECT COUNT(*) FROM src").anzeigen()
// +--------+
// |Anzahl(1)|
// +--------+
// | 500 |
// +--------+
// Die Ergebnisse von SQL-Abfragen sind selbst DataFrames und unterstützen alle normalen Funktionen.
val sqlDF = sql("SELECT-Schlüssel, Wert FROM src WHERE-Schlüssel < 10 ORDER BY-Schlüssel")
// Die Elemente in DataFrames sind vom Typ „Row“, wodurch Sie auf jede Spalte per Ordinalzahl zugreifen können.
val stringsDS = sqlDF.map {
Fall Zeile (Schlüssel: Int, Wert: String) => s"Schlüssel: $Schlüssel, Wert: $Wert"
}
stringsDS.show()
// +--------------------+
// | Wert|
// +--------------------+
// |Schlüssel: 0, Wert: val_0|
// |Schlüssel: 0, Wert: val_0|
// |Schlüssel: 0, Wert: val_0|
// ...
// Sie können DataFrames auch verwenden, um temporäre Ansichten innerhalb einer SparkSession zu erstellen.
val recordsDF = spark.createDataFrame((1 bis 100).map(i => Record(i, s"val_$i")))
recordsDF.createOrReplaceTempView("Aufzeichnungen")
// Abfragen können dann DataFrame-Daten mit in Hive gespeicherten Daten verknüpfen.
sql("SELECT * FROM Datensätze r JOIN src s ON r.key = s.key").show()
// +---+------+---+------+
// |Schlüssel| Wert|Schlüssel| Wert|
// +---+------+---+------+
// | 2| Wert_2| 2| Wert_2|
// | 4| Wert_4| 4| Wert_4|
// | 5| Wert_5| 5| Wert_5|
// ...

spark.sql.warehouse.dir=

$ bin/spark-shell --master spark://hadoop001:7077 --jars mysql-connector-java-5.1.27-bin.jar

{"name":"Michael"}
{"Name":"Andy", "Alter":30}
{"Name":"Justin", "Alter":19}
// Primitive Typen (Int, String, etc.) und Produkttypen (Case-Klassen) Encoder sind
// wird unterstützt, indem dies beim Erstellen eines Datensatzes importiert wird.
importiere spark.implicits._
// Der Pfad zeigt auf einen JSON-Datensatz.
// Der Pfad kann entweder eine einzelne Textdatei oder ein Verzeichnis sein, in dem Textdateien gespeichert sind
val Pfad = "Beispiele/Quelle/Haupt/Ressourcen/Personen.json"
val peopleDF = spark.read.json(Pfad)
// Das abgeleitete Schema kann mit der Methode printSchema() visualisiert werden
peopleDF.printSchema()
//Wurzel
// |-- Alter: lang (nullwertig = wahr)
// |-- Name: Zeichenfolge (nullwertfähig = wahr)
// Erstellt eine temporäre Ansicht unter Verwendung des DataFrame
peopleDF.createOrReplaceTempView("Menschen")
// SQL-Anweisungen können mithilfe der von Spark bereitgestellten SQL-Methoden ausgeführt werden
val teenagerNamesDF = spark.sql("WÄHLEN SIE NAMEN AUS PERSONEN MIT EINEM ALTER ZWISCHEN 13 UND 19")
teenagerNamesDF.show()
// +------+
// | Name|
// +------+
// |Justin|
// +------+
// Alternativ kann ein DataFrame für einen JSON-Datensatz erstellt werden, der dargestellt wird durch
// ein Dataset[String], das ein JSON-Objekt pro String speichert
Wert otherPeopleDataset = spark.createDataset(
"""{"Name":"Yin","Adresse":{"Stadt":"Columbus","Staat":"Ohio"}}""" :: Nil)
val anderePersonen = spark.read.json(anderePersonenDataset)
anderePersonen.show()
// +---------------+----+
// | Adresse|Name|
// +---------------+----+
// |[Columbus,Ohio]| Yin|
// +---------------+----+

$ bin/spark-shell --master spark://hadoop001:7077 --jars mysql-connector-java-5.1.27-bin.jar
// Hinweis: Das Laden und Speichern von JDBC kann entweder über die Methoden load/save oder jdbc erfolgen.
// Daten aus einer JDBC-Quelle laden
val jdbcDF = spark.read.format("jdbc").option("url", "jdbc:mysql://hadoop001:3306/rdd").option("dbtable", " rddtable").option("Benutzer", "root").option("Passwort", "hive").load()
val connectionProperties = neue Eigenschaften()
connectionProperties.put("Benutzer", "root")
connectionProperties.put("Passwort", "Hive")
val jdbcDF2 = spark.read
.jdbc("jdbc:mysql://hadoop001:3306/rdd", "rddtable", Verbindungseigenschaften)
// Daten in einer JDBC-Quelle speichern
jdbcDF.schreiben
.format("jdbc")
.option("url", "jdbc:mysql://hadoop001:3306/rdd")
.option("dbtable", "rddtable2")
.option("Benutzer", "root")
.option("Passwort", "Hive")
.speichern()
jdbcDF2.schreiben
.jdbc("jdbc:mysql://hadoop001:3306/mysql", "db", Verbindungseigenschaften)
// Datentypen für Tabellenspalten beim Schreiben festlegen
jdbcDF.schreiben
.option("createTableColumnTypes", "Name CHAR(64), Kommentare VARCHAR(1024)")
.jdbc("jdbc:mysql://hadoop001:3306/mysql", "db", Verbindungseigenschaften)