SparkSQL: intra-SparkSQL-Anwendungstabelle Registrierung

Question

Kontext. Ich habe Dutzende von SQL-Abfragen in separaten Dateien gespeichert. Zu Benchmarkzwecken habe ich eine Anwendung erstellt, die jede dieser Abfragedateien durchläuft und an eine eigenständige Spark-Anwendung übergibt. Letzterer analysiert zuerst die Abfrage, extrahiert die verwendeten Tabellen, registriert sie (unter Verwendung von: registerTempTable() in Spark < 2 und createOrReplaceTempView() in Spark 2) und führt die Abfrage effektiv aus (spark.sql()).

Herausforderung. Da die Tabellen Registrierung kann sehr zeitaufwendig sein, würde Ich mag träge registrieren um die Tabellen , dh nur einmal wenn sie ersten verwendet werden, und das in Form von Metadaten halten, die in den nachfolgenden Abfragen ohne weiteres verwendet werden können, ohne die Notwendigkeit, die Tabellen bei jeder Abfrage neu zu registrieren. Es ist eine Art Intra-Job-Caching, aber keine der Caching-Optionen, die Spark bietet (Tabellen-Caching), soweit ich weiß.

Ist das möglich? Wenn nicht, kann jemand einen anderen Ansatz vorschlagen, um das gleiche Ziel zu erreichen (indem man durch getrennte Abfragedateien iteriert und eine abfragende Spark-Anwendung ausführt, ohne die Tabellen zu registrieren, die bereits zuvor registriert wurden).

Answer 1

Im Allgemeinen sollte die Registrierung einer Tabelle nicht lange dauern (außer, wenn Sie viele Dateien haben, kann es einige Zeit dauern, die Liste der Dateiquellen zu erstellen). Es gibt im Grunde nur dem Datenrahmen einen Namen. Was Zeit kosten würde, ist das Lesen des Datenrahmens von der Festplatte.

Die grundlegende Frage ist also, wie wird der Datenrahmen (Tabellen) auf die Festplatte geschrieben. Wenn es als eine große Anzahl von kleinen Dateien oder ein Dateiformat, das langsam ist (z. B. CSV) geschrieben wird, kann dies einige Zeit dauern (viele Dateien brauchen Zeit, um die Dateiliste zu generieren und ein "langsames" Dateiformat bedeutet die das tatsächliche Lesen ist langsam).

Das erste, was Sie versuchen können, ist, Ihre Daten zu lesen und erneut zu speichern. Nehmen wir als Beispiel, dass Sie eine große Anzahl von CSV-Dateien in einem Pfad haben. jetzt

df = spark.read.csv("path/*.csv") 

, dass Sie einen Datenrahmen haben, können Sie es ändern weniger Dateien haben und ein besseres Format wie zu verwenden: Sie können so etwas wie tun

df.coalesce(100).write.parquet("newPath") 

Wenn die oben nicht genug ist, und Ihr Cluster ist groß genug, um alles zwischenzuspeichern, Sie könnten alles in einen einzigen Job stecken, alle Tabellen in allen Abfragen durchsuchen, alle davon registrieren und zwischenspeichern. Führen Sie dann nacheinander Ihre SQL-Abfragen aus (und geben Sie diese einzeln ein).

Wenn all dies fehlschlägt, können Sie versuchen, etwas wie alluxio (http://www.alluxio.org/) zu verwenden, um ein Dateisystem im Speicher zu erstellen und zu versuchen, daraus zu lesen.