Abfragen von Amazon Redshift mit Azure Databricks

Sie können Tabellen aus Amazon Redshift mit Azure Databricks lesen und schreiben.

Die Databricks Redshift-Datenquelle verwendet Amazon S3, um Daten effizient in und aus Redshift zu übertragen und nutzt JDBC, um automatisch die entsprechenden COPY- und UNLOAD-Befehle auf Redshift auszulösen.

Verwendung

Die folgenden Beispiele veranschaulichen die Verbindung mit dem Redshift-Treiber. Ersetzen Sie die url Parameterwerte, falls Sie den PostgreSQL JDBC-Treiber verwenden.

Nachdem Sie Ihre AWS-Anmeldeinformationen konfiguriert haben, können Sie die Datenquelle mit der Spark-Datenquellen-API in Python, SQL, R oder Scala verwenden.

Python

# Read data from a table using Databricks Runtime 10.4 LTS and below
df = (spark.read
  .format("redshift")
  .option("dbtable", table_name)
  .option("tempdir", "s3a://<bucket>/<directory-path>")
  .option("url", "jdbc:redshift://<database-host-url>")
  .option("user", username)
  .option("password", password)
  .option("forward_spark_s3_credentials", True)
  .load()
)

# Read data from a table using Databricks Runtime 11.3 LTS and above
df = (spark.read
  .format("redshift")
  .option("host", "hostname")
  .option("port", "port") # Optional - will use default port 5439 if not specified.
  .option("user", "username")
  .option("password", "password")
  .option("database", "database-name")
  .option("dbtable", "schema-name.table-name") # if schema-name is not specified, default to "public".
  .option("tempdir", "s3a://<bucket>/<directory-path>")
  .option("forward_spark_s3_credentials", True)
  .load()
)

# Read data from a query
df = (spark.read
  .format("redshift")
  .option("query", "select x, count(*) <your-table-name> group by x")
  .option("tempdir", "s3a://<bucket>/<directory-path>")
  .option("url", "jdbc:redshift://<database-host-url>")
  .option("user", username)
  .option("password", password)
  .option("forward_spark_s3_credentials", True)
  .load()
)

# After you have applied transformations to the data, you can use
# the data source API to write the data back to another table

# Write back to a table
(df.write
  .format("redshift")
  .option("dbtable", table_name)
  .option("tempdir", "s3a://<bucket>/<directory-path>")
  .option("url", "jdbc:redshift://<database-host-url>")
  .option("user", username)
  .option("password", password)
  .mode("error")
  .save()
)

# Write back to a table using IAM Role based authentication
(df.write
  .format("redshift")
  .option("dbtable", table_name)
  .option("tempdir", "s3a://<bucket>/<directory-path>")
  .option("url", "jdbc:redshift://<database-host-url>")
  .option("user", username)
  .option("password", password)
  .option("aws_iam_role", "arn:aws:iam::123456789000:role/redshift_iam_role")
  .mode("error")
  .save()
)

SQL

Lesen von Daten mithilfe von SQL für Databricks Runtime 10.4 LTS und darunter:

DROP TABLE IF EXISTS redshift_table;
CREATE TABLE redshift_table
USING redshift
OPTIONS (
  dbtable '<table-name>',
  tempdir 's3a://<bucket>/<directory-path>',
  url 'jdbc:redshift://<database-host-url>',
  user '<username>',
  password '<password>',
  forward_spark_s3_credentials 'true'
);
SELECT * FROM redshift_table;

Lesen von Daten mithilfe von SQL für Databricks Runtime 11.3 LTS und höher:

DROP TABLE IF EXISTS redshift_table;
CREATE TABLE redshift_table
USING redshift
OPTIONS (
  host '<hostname>',
  port '<port>', /* Optional - will use default port 5439 if not specified. *./
  user '<username>',
  password '<password>',
  database '<database-name>'
  dbtable '<schema-name>.<table-name>', /* if schema-name not provided, default to "public". */
  tempdir 's3a://<bucket>/<directory-path>',
  forward_spark_s3_credentials 'true'
);
SELECT * FROM redshift_table;

Schreiben von Daten mit SQL:

DROP TABLE IF EXISTS redshift_table;
CREATE TABLE redshift_table_new
USING redshift
OPTIONS (
  dbtable '<new-table-name>',
  tempdir 's3a://<bucket>/<directory-path>',
  url 'jdbc:redshift://<database-host-url>',
  user '<username>',
  password '<password>',
  forward_spark_s3_credentials 'true'
) AS
SELECT * FROM table_name;

Die SQL-API unterstützt nur die Erstellung neuer Tabellen und nicht das Überschreiben oder Anfügen.

R

Lesen von Daten mithilfe von R in Databricks Runtime 10.4 LTS und früher:

df <- read.df(
   NULL,
   "com.databricks.spark.redshift",
   tempdir = "s3a://<your-bucket>/<your-directory-path>",
   dbtable = "<your-table-name>",
   url = "jdbc:redshift://<the-rest-of-the-connection-string>")

Lesen von Daten mit R on Databricks Runtime 11.3 LTS und höher:

df <- read.df(
  NULL,
  "redshift",
  host = "hostname",
  port = "port",
  user = "username",
  password = "password",
  database = "database-name",
  dbtable = "schema-name.table-name",
  tempdir = "s3a://<your-bucket>/<your-directory-path>",
  forward_spark_s3_credentials = "true",
  dbtable = "<your-table-name>")

Skala

// Read data from a table using Databricks Runtime 10.4 LTS and below
val df = spark.read
  .format("redshift")
  .option("dbtable", table_name)
  .option("tempdir", "s3a://<bucket>/<directory-path>")
  .option("url", "jdbc:redshift://<database-host-url>")
  .option("user", username)
  .option("password", password)
  .option("forward_spark_s3_credentials", True)
  .load()

// Read data from a table using Databricks Runtime 11.3 LTS and above
val df = spark.read
  .format("redshift")
  .option("host", "hostname")
  .option("port", "port") /* Optional - will use default port 5439 if not specified. */
  .option("user", "username")
  .option("password", "password")
  .option("database", "database-name")
  .option("dbtable", "schema-name.table-name") /* if schema-name is not specified, default to "public". */
  .option("tempdir", "s3a://<bucket>/<directory-path>")
  .option("forward_spark_s3_credentials", true)
  .load()

// Read data from a query
val df = spark.read
  .format("redshift")
  .option("query", "select x, count(*) <your-table-name> group by x")
  .option("tempdir", "s3a://<bucket>/<directory-path>")
  .option("url", "jdbc:redshift://<database-host-url>")
  .option("user", username)
  .option("password", password)
  .option("forward_spark_s3_credentials", True)
  .load()

// After you have applied transformations to the data, you can use
// the data source API to write the data back to another table

// Write back to a table
df.write
  .format("redshift")
  .option("dbtable", table_name)
  .option("tempdir", "s3a://<bucket>/<directory-path>")
  .option("url", "jdbc:redshift://<database-host-url>")
  .option("user", username)
  .option("password", password)
  .mode("error")
  .save()

// Write back to a table using IAM Role based authentication
df.write
  .format("redshift")
  .option("dbtable", table_name)
  .option("tempdir", "s3a://<bucket>/<directory-path>")
  .option("url", "jdbc:redshift://<database-host-url>")
  .option("user", username)
  .option("password", password)
  .option("aws_iam_role", "arn:aws:iam::123456789000:role/redshift_iam_role")
  .mode("error")
  .save()

Empfehlungen für das Arbeiten mit Redshift

Die Abfrageausführung kann große Datenmengen in S3 extrahieren. Wenn Sie mehrere Abfragen für dieselben Daten in Redshift durchführen möchten, empfiehlt Databricks, die extrahierten Daten mithilfe von Delta Lake zu speichern.

Konfiguration

Authentifizieren bei S3 und Redshift

Die Datenquelle umfasst mehrere Netzwerkverbindungen, die im folgenden Diagramm veranschaulicht werden:

                            ┌───────┐
       ┌───────────────────>│  S3   │<─────────────────┐
       │    IAM or keys     └───────┘    IAM or keys   │
       │                        ^                      │
       │                        │ IAM or keys          │
       v                        v               ┌──────v────┐
┌────────────┐            ┌───────────┐         │┌──────────┴┐
│  Redshift  │            │  Spark    │         ││   Spark   │
│            │<──────────>│  Driver   │<────────>| Executors │
└────────────┘            └───────────┘          └───────────┘
               JDBC with                  Configured
               username /                     in
               password                     Spark
        (SSL enabled by default)

Die Datenquelle liest und schreibt Daten in S3, wenn Daten an/aus Redshift übertragen werden. Daher erfordert es AWS-Anmeldeinformationen mit Lese- und Schreibzugriff auf einen S3-Bucket (angegeben mithilfe des tempdir Konfigurationsparameters).

In den folgenden Abschnitten werden die Konfigurationsoptionen für die Authentifizierung der einzelnen Verbindungen beschrieben:

• Spark-Treiber zu Redshift
• Spark to S3
• Redshift zu S3

Spark-Treiber zu Redshift

Der Spark-Treiber verbindet sich über JDBC mit Redshift und verwendet dabei einen Benutzernamen und ein Passwort. Redshift unterstützt nicht die Verwendung von IAM-Rollen zum Authentifizieren dieser Verbindung. Standardmäßig verwendet diese Verbindung SSL-Verschlüsselung; weitere Informationen finden Sie unter Verschlüsselung.

Spark zu S3

S3 fungiert als Vermittler zum Speichern von Massendaten beim Lesen oder Schreiben in Redshift. Spark stellt eine Verbindung zu S3 sowohl über die Hadoop FileSystem-Schnittstellen als auch direkt über den S3-Client des Amazon Java SDK her.

• Festlegen von Schlüsseln in Hadoop conf: Sie können AWS-Schlüssel mithilfe von Hadoop-Konfigurationseigenschaften angeben. Wenn Ihre tempdir Konfiguration auf ein s3a:// Dateisystem verweist, können Sie die fs.s3a.access.key und fs.s3a.secret.key Eigenschaften in einer Hadoop-XML-Konfigurationsdatei festlegen oder sc.hadoopConfiguration.set() aufrufen, um die globale Hadoop-Konfiguration von Spark zu konfigurieren. Wenn Sie ein s3n:// Dateisystem verwenden, können Sie die Legacykonfigurationsschlüssel bereitstellen, wie im folgenden Beispiel gezeigt.

  Skala

  Wenn Sie beispielsweise das s3a Dateisystem verwenden, fügen Sie Folgendes hinzu:

  sc.hadoopConfiguration.set("fs.s3a.access.key", "<your-access-key-id>")
  sc.hadoopConfiguration.set("fs.s3a.secret.key", "<your-secret-key>")
  

  Fügen Sie für das Legacydateisystem s3n Folgendes hinzu:

  sc.hadoopConfiguration.set("fs.s3n.awsAccessKeyId", "<your-access-key-id>")
  sc.hadoopConfiguration.set("fs.s3n.awsSecretAccessKey", "<your-secret-key>")
  

  Python

  Der folgende Befehl basiert auf einigen Spark-Internen, sollte jedoch mit allen PySpark-Versionen arbeiten und ist in Zukunft unwahrscheinlich, dass sich dies ändert:

    sc._jsc.hadoopConfiguration().set("fs.s3a.access.key", "<your-access-key-id>")
    sc._jsc.hadoopConfiguration().set("fs.s3a.secret.key", "<your-secret-key>")
  

Redshift zu S3

Legen Sie die forward_spark_s3_credentials Option auf true fest, um die von Spark zum Herstellen einer Verbindung mit S3 über JDBC verwendeten AWS-Schlüsselanmeldeinformationen automatisch an Redshift weiterzuleiten. Die JDBC-Abfrage bettet diese Anmeldeinformationen ein, sodass Databricks dringend empfiehlt, dass Sie die SSL-Verschlüsselung der JDBC-Verbindung aktivieren.

Verschlüsselung

• Absicherung von JDBC: Sofern keine SSL-bezogenen Einstellungen in der JDBC-URL vorhanden sind, aktiviert die Datenquelle standardmäßig SSL-Verschlüsselung und überprüft außerdem, ob der Redshift-Server vertrauenswürdig ist (d. h.sslmode=verify-full). Dazu wird automatisch ein Serverzertifikat von den Amazon-Servern heruntergeladen, wenn es zum ersten Mal benötigt wird. Falls dies fehlschlägt, wird eine vorab gebündelte Zertifikatdatei als Fallback verwendet. Dies gilt sowohl für die Redshift- als auch für die PostgreSQL JDBC-Treiber.

  Falls Probleme mit diesem Feature auftreten, oder falls Sie SSL einfach deaktivieren möchten, können Sie .option("autoenablessl", "false") auf Ihrem DataFrameReader oder Ihrem DataFrameWriter aufrufen.

  Wenn Sie benutzerdefinierte SSL-bezogene Einstellungen angeben möchten, können Sie die Anweisungen in der Redshift-Dokumentation befolgen: Verwenden von SSL- und Serverzertifikaten in Java und JDBC-Treiberkonfigurationsoptionen. Alle SSL-bezogenen Optionen, die im JDBC url mit der Datenquelle verwendet werden, haben Vorrang (d. h. die automatische Konfiguration wird nicht ausgelöst).

• Verschlüsseln von UNLOAD-Daten, die in S3 gespeichert sind (Beim Lesen von Redshift gespeicherte Daten): Laut redshift-Dokumentation zum Entladen von Daten nach S3 verschlüsselt UNLOAD automatisch Datendateien mithilfe der serverseitigen Amazon S3-Verschlüsselung (SSE-S3)."

  Redshift unterstützt auch die clientseitige Verschlüsselung mit einem benutzerdefinierten Schlüssel (siehe: Entladen verschlüsselter Datendateien), die Datenquelle verfügt jedoch nicht über die Möglichkeit, den erforderlichen symmetrischen Schlüssel anzugeben.

• Verschlüsseln von COPY-Daten, die in S3 gespeichert sind (Beim Schreiben in Redshift gespeicherte Daten): Laut redshift-Dokumentation zum Laden verschlüsselter Datendateien von Amazon S3:

Sie können den COPY Befehl verwenden, um Datendateien zu laden, die mit serverseitiger Verschlüsselung mit von AWS verwalteten Verschlüsselungsschlüsseln (SSE-S3 oder SSE-KMS), clientseitiger Verschlüsselung oder beides in Amazon S3 hochgeladen wurden. COPY unterstützt keine serverseitige Amazon S3-Verschlüsselung mit einem vom Kunden bereitgestellten Schlüssel (SSE-C).

Parameter

Die in Spark SQL bereitgestellte Parameterzuordnung oder -optionen unterstützen die folgenden Einstellungen:

Zusätzliche Konfigurationsoptionen

Konfigurieren der maximalen Größe von Zeichenfolgenspalten

Beim Erstellen von Redshift-Tabellen besteht das Standardverhalten darin, TEXT Spalten für Zeichenfolgenspalten zu erstellen. Rotshift speichert TEXT Spalten als VARCHAR(256), sodass diese Spalten eine maximale Größe von 256 Zeichen (Quelle) aufweisen.

Um größere Spalten zu unterstützen, können Sie das maxlength Spaltenmetadatenfeld verwenden, um die maximale Länge einzelner Zeichenfolgenspalten anzugeben. Dies ist auch hilfreich für die Implementierung von leistungssparenden Leistungsoptimierungen, indem Spalten mit einer geringeren maximalen Länge als der Standard deklariert werden.

Python

df = ... # the dataframe you'll want to write to Redshift

# Specify the custom width of each column
columnLengthMap = {
  "language_code": 2,
  "country_code": 2,
  "url": 2083,
}

# Apply each column metadata customization
for (colName, length) in columnLengthMap.iteritems():
  metadata = {'maxlength': length}
  df = df.withColumn(colName, df[colName].alias(colName, metadata=metadata))

df.write \
  .format("com.databricks.spark.redshift") \
  .option("url", jdbcURL) \
  .option("tempdir", s3TempDirectory) \
  .option("dbtable", sessionTable) \
  .save()

Skala

Hier ist ein Beispiel für das Aktualisieren der Metadatenfelder mehrerer Spalten mithilfe der Scala-API von Spark:

import org.apache.spark.sql.types.MetadataBuilder

// Specify the custom width of each column
val columnLengthMap = Map(
  "language_code" -> 2,
  "country_code" -> 2,
  "url" -> 2083
)

var df = ... // the dataframe you'll want to write to Redshift

// Apply each column metadata customization
columnLengthMap.foreach { case (colName, length) =>
  val metadata = new MetadataBuilder().putLong("maxlength", length).build()
  df = df.withColumn(colName, df(colName).as(colName, metadata))
}

df.write
  .format("com.databricks.spark.redshift")
  .option("url", jdbcURL)
  .option("tempdir", s3TempDirectory)
  .option("dbtable", sessionTable)
.save()

Festlegen eines benutzerdefinierten Spaltentyps

Wenn Sie einen Spaltentyp manuell festlegen müssen, können Sie die redshift_type Spaltenmetadaten verwenden. Wenn Sie z. B. den Spark SQL Schema -> Redshift SQL-Typabgleicher überschreiben möchten, um einen benutzerdefinierten Spaltentyp zuzuweisen, können Sie wie folgt vorgehen:

Python

# Specify the custom type of each column
columnTypeMap = {
  "language_code": "CHAR(2)",
  "country_code": "CHAR(2)",
  "url": "BPCHAR(111)",
}

df = ... # the dataframe you'll want to write to Redshift

# Apply each column metadata customization
for colName, colType in columnTypeMap.items():
  metadata = {'redshift_type': colType}
  df = df.withColumn(colName, df[colName].alias(colName, metadata=metadata))

Skala

import org.apache.spark.sql.types.MetadataBuilder

// Specify the custom type of each column
val columnTypeMap = Map(
  "language_code" -> "CHAR(2)",
  "country_code" -> "CHAR(2)",
  "url" -> "BPCHAR(111)"
)

var df = ... // the dataframe you'll want to write to Redshift

// Apply each column metadata customization
columnTypeMap.foreach { case (colName, colType) =>
  val metadata = new MetadataBuilder().putString("redshift_type", colType).build()
  df = df.withColumn(colName, df(colName).as(colName, metadata))
}

Konfigurieren der Spaltencodierung

Verwenden Sie beim Erstellen einer Tabelle das encoding Spaltenmetadatenfeld, um eine Komprimierungscodierung für jede Spalte anzugeben (siehe Amazon-Dokumente für verfügbare Codierungen).

Festlegen von Beschreibungen für Spalten

Mit "Redshift" können Spalten Beschreibungen angefügt werden, die in den meisten Abfragetools angezeigt werden sollen (mithilfe des COMMENT Befehls). Sie können das description Spaltenmetadatenfeld so festlegen, dass eine Beschreibung für einzelne Spalten angegeben wird.

Abfrage-Pushdown in Redshift

Der Spark-Optimierer verschiebt die folgenden Operatoren nach unten in Redshift:

• Filter
• Project
• Sort
• Limit
• Aggregation
• Join

Innerhalb Project und Filter, es unterstützt die folgenden Ausdrücke:

• Die meisten booleschen Logikoperatoren
• Vergleiche
• Grundlegende arithmetische Operationen
• Numerische und Zeichenfolgenumwandlungen
• Die meisten Zeichenfolgenfunktionen
• Skalare Unterabfragen, wenn sie vollständig in Redshift gepusht werden können.

Innerhalb von Aggregation unterstützt er die folgenden Aggregationsfunktionen:

• AVG
• COUNT
• MAX
• MIN
• SUM
• STDDEV_SAMP
• STDDEV_POP
• VAR_SAMP
• VAR_POP

in Kombination mit der DISTINCT Klausel, sofern zutreffend.

Innerhalb von Join unterstützt er die folgenden Arten von Verknüpfungen:

• INNER JOIN
• LEFT OUTER JOIN
• RIGHT OUTER JOIN
• LEFT SEMI JOIN
• LEFT ANTI JOIN
• Unterabfragen, die der Optimierer neu in Join schreibt, z. B. WHERE EXISTS, WHERE NOT EXISTS

Der Pushdown ist u. U. bei Abfragen mit LIMIT am nützlichsten. Eine Abfrage, wie z. B. SELECT * FROM large_redshift_table LIMIT 10 könnte sehr lange dauern, da die gesamte Tabelle zunächst per UNLOAD als Zwischenergebnis in S3 entladen würde. Mit dem Pushdown wird LIMIT in Redshift ausgeführt. Bei Abfragen mit Aggregationen hilft auch der Pushdown der Aggregation in Redshift bei der Reduzierung der Datenmenge, die übertragen werden muss.

Der Abfrage-Pushdown in Redshift ist standardmäßig aktiviert. Die Funktion kann deaktiviert werden, indem spark.databricks.redshift.pushdown auf false gesetzt wird. Auch wenn sie deaktiviert sind, pusht Spark weiterhin Filter nach unten und führt die Spaltenauslöschung in Redshift durch.

Installation des Redshift-Treibers

Die Redshift-Datenquelle erfordert auch einen Redshift-kompatiblen JDBC-Treiber. Da Redshift auf dem PostgreSQL-Datenbanksystem basiert, können Sie den PostgreSQL-JDBC-Treiber verwenden, der im Lieferumfang der Databricks Runtime enthalten ist, oder den von Amazon empfohlenen Redshift JDBC-Treiber. Es ist keine Installation erforderlich, um den PostgreSQL JDBC-Treiber zu verwenden. Die Version des in jeder Databricks Runtime-Version enthaltenen PostgreSQL JDBC-Treibers ist in den Databricks Runtime-Versionshinweisen aufgeführt.

So installieren Sie den Redshift JDBC-Treiber manuell:

1. Laden Sie den Treiber von Amazon herunter.
2. Laden Sie den Treiber in Ihren Azure Databricks-Arbeitsbereich hoch. Siehe Installieren von Bibliotheken.
3. Installieren Sie die Bibliothek auf Ihrem Cluster.

Transaktionsgarantien

In diesem Abschnitt werden die Transaktionsgarantien der Redshift-Datenquelle für Spark beschrieben.

Allgemeiner Hintergrund für Redshift- und S3-Eigenschaften

Allgemeine Informationen zu Redshift-Transaktionsgarantien finden Sie im Kapitel "Managing Concurrent Write Operations " in der Redshift-Dokumentation. Kurz gesagt, Redshift bietet serialisierbare Isolation gemäß der Dokumentation für den Befehl Redshift BEGIN :

[obwohl] Sie alle vier Transaktionsisolationsstufen verwenden können, verarbeitet Amazon Redshift alle Isolationsstufen als serialisierbar.

Gemäß der Redshift-Dokumentation:

Amazon Redshift unterstützt ein standardmäßiges automatisches Commit-Verhalten , bei dem jeder separat ausgeführte SQL-Befehl einzeln committ.

Daher sind einzelne Befehle wie COPY und UNLOAD atomar und transaktional, während explizite BEGIN und END nur erforderlich sein sollten, um die Atomarität mehrerer Befehle oder Abfragen zu erzwingen.

Beim Lesen von und Schreiben in Redshift liest und schreibt die Datenquelle Daten in S3. Sowohl Spark als auch Redshift erzeugen partitionierte Ausgabe und speichern sie in mehreren Dateien in S3. Gemäß der Dokumentation zum Amazon S3-Datenkonsistenzmodell sind S3-Bucketauflistungsvorgänge letztendlich konsistent, sodass bei den Dateien besondere Maßnahmen ergriffen werden müssen, um fehlende oder unvollständige Daten aufgrund dieser Quelle letztendlicher Konsistenz zu vermeiden.

Garantien der Redshift-Datenquelle für Spark

Anfügen an eine vorhandene Tabelle

Beim Einfügen von Zeilen in Redshift verwendet die Datenquelle den BEFEHL COPY und gibt Manifeste an, die vor bestimmten letztendlich konsistenten S3-Vorgängen schützen. spark-redshift Daher weisen Anfüge an vorhandene Tabellen dieselben atomischen und transaktionsalen Eigenschaften wie normale Redshift-Befehle COPY auf.

Erstellen einer neuen Tabelle (SaveMode.CreateIfNotExists)

Das Erstellen einer neuen Tabelle ist ein zweistufiger Prozess, der aus einem CREATE TABLE Befehl gefolgt von einem COPY-Befehl besteht, um den anfänglichen Satz von Zeilen anzufügen. Beide Vorgänge werden in derselben Transaktion ausgeführt.

Überschreiben einer vorhandenen Tabelle

Standardmäßig verwendet die Datenquelle Transaktionen, um Überschreibungen auszuführen, die durch Löschen der Zieltabelle, das Erstellen einer neuen leeren Tabelle und das Anfügen von Zeilen an sie implementiert werden.

Wenn die veraltete usestagingtable Einstellung auf false festgelegt ist, führt die Datenquelle den DELETE TABLE Befehl aus, bevor sie Zeilen an die neue Tabelle anhängt. Dies opfert die Atomizität des Überschreibvorgangs, verringert jedoch den benötigten Stagingbereich, den Redshift während des Überschreibens benötigt.

Redshift-Tabellenabfrage

Abfragen verwenden den Redshift-Befehl UNLOAD, um eine Abfrage auszuführen und die Ergebnisse in S3 zu speichern. Manifeste werden verwendet, um gegen bestimmte eventuell konsistente S3-Vorgänge zu schützen. Daher sollten Abfragen aus der Redshift-Datenquelle für Spark dieselben Konsistenzeigenschaften wie normale Redshift-Abfragen aufweisen.

Häufige Probleme und Lösungen

S3 Bucket- und Redshift-Cluster befinden sich in verschiedenen AWS-Regionen

Standardmäßig funktionieren S3 <-> Redshift-Kopien nicht, wenn sich der S3-Bucket und der Redshift-Cluster in verschiedenen AWS-Regionen befinden.

Wenn Sie versuchen, eine Redshift-Tabelle zu lesen, wenn sich der S3-Bucket in einem anderen Bereich befindet, wird möglicherweise ein Fehler angezeigt, z. B.:

ERROR: S3ServiceException:The S3 bucket addressed by the query is in a different region from this cluster.,Status 301,Error PermanentRedirect.

Ebenso kann der Versuch, mithilfe eines S3-Buckets in einem anderen Bereich in Redshift zu schreiben, den folgenden Fehler verursachen:

error:  Problem reading manifest file - S3ServiceException:The S3 bucket addressed by the query is in a different region from this cluster.,Status 301,Error PermanentRedirect

• Schreibt: Der Befehl "Redshift COPY" unterstützt die explizite Spezifikation des S3-Bucketbereichs, sodass Schreibvorgänge in Redshift in diesen Fällen ordnungsgemäß durchgeführt werden können, indem Sie region 'the-region-name' zur extracopyoptions Einstellung hinzufügen. Verwenden Sie beispielsweise mit einem Bucket in der Region USA, Osten (Virginia) und der Skala-API:

  .option("extracopyoptions", "region 'us-east-1'")
  

  Alternativ können Sie die awsregion Einstellung verwenden:

  .option("awsregion", "us-east-1")
  

• Liest: Der Befehl Redshift UNLOAD unterstützt auch die explizite Spezifikation der S3-Bucket-Region. Sie können Lesevorgänge ordnungsgemäß vornehmen, indem Sie der awsregion-Einstellung die Region hinzufügen:

  .option("awsregion", "us-east-1")
  

Authentifizierungsfehler beim Verwenden eines Kennworts mit Sonderzeichen in der JDBC-URL

Wenn Sie den Benutzernamen und das Kennwort als Teil der JDBC-URL angeben und das Kennwort Sonderzeichen wie ;, ?oder & enthält, wird möglicherweise die folgende Ausnahme angezeigt:

java.sql.SQLException: [Amazon](500310) Invalid operation: password authentication failed for user 'xyz'

Dies wird durch Sonderzeichen im Benutzernamen oder Kennwort verursacht, die vom JDBC-Treiber nicht ordnungsgemäß entfernt werden. Stellen Sie sicher, dass Sie den Benutzernamen und das Kennwort mithilfe der entsprechenden DataFrame-Optionen user und password angeben. Weitere Informationen finden Sie unter Parameter.

Lange ausgeführte Spark-Abfrage hängt unbegrenzt, obwohl der entsprechende Redshift-Vorgang abgeschlossen ist.

Wenn Sie große Datenmengen aus und in Redshift lesen oder schreiben, hängt Ihre Spark-Abfrage möglicherweise unbegrenzt, auch wenn auf der AWS Redshift Monitoring-Seite angezeigt wird, dass der entsprechende LOAD- oder UNLOAD-Vorgang abgeschlossen ist und dass Cluster im Leerlauf ist. Dies wird durch die Verbindung zwischen Redshift und Spark verursacht. Um dies zu vermeiden, vergewissern Sie sich, dass das tcpKeepAlive JDBC-Flag aktiviert ist und TCPKeepAliveMinutes auf einen niedrigen Wert festgelegt ist (z. B. 1).

Weitere Informationen finden Sie unter Amazon Redshift JDBC Driver Configuration.

Zeitstempel mit Zeitzonensemantik

Beim Lesen von Daten werden Redshift TIMESTAMP- und TIMESTAMPTZ-Datentypen dem Spark-TimestampType zugeordnet, und ein Wert wird in koordinierte Weltzeit (UTC) konvertiert und als UTC-Zeitstempel gespeichert. Bei einer Redshift TIMESTAMPwird die lokale Zeitzone angenommen, da der Wert keine Zeitzoneninformationen enthält. Beim Schreiben von Daten in eine Redshift-Tabelle wird ein Spark TimestampType dem Redshift-Datentyp TIMESTAMP zugeordnet.

Migrationsleitfaden

Die Datenquelle erfordert jetzt, dass Sie explizit forward_spark_s3_credentials festlegen, bevor Spark S3-Anmeldeinformationen an Redshift weitergeleitet werden. Diese Änderung hat keine Auswirkungen, wenn Sie die aws_iam_role- oder temporary_aws_*-Authentifizierungsmechanismen verwenden. Wenn Sie sich jedoch auf das alte Standardverhalten verlassen haben, müssen Sie jetzt explizit forward_spark_s3_credentials auf true setzen, um Ihren vorherigen Redshift-zu-S3-Authentifizierungsmechanismus weiterhin verwenden zu können. Eine Diskussion der drei Authentifizierungsmechanismen und ihrer Sicherheitskompromisse finden Sie im Abschnitt "Authentifizierung zu S3 und Redshift" dieses Dokuments.