Wie verwende ich die JDBC-Quelle zum Schreiben und Lesen von Daten in (Py) Spark?

69

Das Ziel dieser Frage ist es zu dokumentieren:

  • Schritte zum Lesen und Schreiben von Daten über JDBC-Verbindungen in PySpark

  • mögliche Probleme mit JDBC-Quellen und bekannten Lösungen

Mit kleinen Änderungen sollten diese Methoden mit anderen unterstützten Sprachen wie Scala und R funktionieren.

null323
quelle

Antworten:

122

Daten schreiben

  1. Fügen Sie den entsprechenden JDBC-Treiber hinzu, wenn Sie die Anwendung senden oder die Shell starten. Sie können zum Beispiel verwenden --packages:

    bin/pyspark --packages group:name:version  
    

    oder kombinieren driver-class-pathundjars

    bin/pyspark --driver-class-path $PATH_TO_DRIVER_JAR --jars $PATH_TO_DRIVER_JAR
    

    Diese Eigenschaften können auch mithilfe der PYSPARK_SUBMIT_ARGSUmgebungsvariablen festgelegt werden, bevor die JVM-Instanz gestartet wurde, oder mithilfe conf/spark-defaults.confvon set spark.jars.packagesoder spark.jars/ spark.driver.extraClassPath.

  2. Wählen Sie den gewünschten Modus. Der Spark JDBC-Writer unterstützt folgende Modi:

    • append: Inhalt dieser Klasse DataFramean vorhandene Daten anhängen .
    • overwrite: Vorhandene Daten überschreiben.
    • ignore: Ignorieren Sie diesen Vorgang stillschweigend, wenn bereits Daten vorhanden sind.
    • error (Standardfall): Löst eine Ausnahme aus, wenn bereits Daten vorhanden sind.

    Upserts oder andere feinkörnige Modifikationen werden nicht unterstützt

    mode = ...
    
  3. Bereiten Sie den JDBC-URI vor, zum Beispiel:

    # You can encode credentials in URI or pass
    # separately using properties argument
    # of jdbc method or options
    
    url = "jdbc:postgresql://localhost/foobar"
    
  4. (Optional) Erstellen Sie ein Wörterbuch mit JDBC-Argumenten.

    properties = {
        "user": "foo",
        "password": "bar"
    }
    

    properties/ optionskann auch verwendet werden, um unterstützte JDBC-Verbindungseigenschaften festzulegen .

  5. Verwenden DataFrame.write.jdbc

    df.write.jdbc(url=url, table="baz", mode=mode, properties=properties)
    

    um die Daten zu speichern (siehe pyspark.sql.DataFrameWriterfür Details).

Bekannte Probleme :

  • Ein geeigneter Treiber kann nicht gefunden werden, wenn der Treiber mit --packages(java.sql.SQLException: No suitable driver found for jdbc: ... ) aufgenommen wurde.

    Angenommen, es gibt keine Nichtübereinstimmung der Treiberversion, um dies zu beheben, können Sie der driverKlasse eine hinzufügen properties. Zum Beispiel:

    properties = {
        ...
        "driver": "org.postgresql.Driver"
    }
    
  • Verwendung df.write.format("jdbc").options(...).save()kann führen zu:

    java.lang.RuntimeException: org.apache.spark.sql.execution.datasources.jdbc.DefaultSource erlaubt das Erstellen einer Tabelle nicht als Auswahl.

    Lösung unbekannt.

  • In Pyspark 1.3 können Sie versuchen, die Java-Methode direkt aufzurufen:

    df._jdf.insertIntoJDBC(url, "baz", True)
    

Daten lesen

  1. Befolgen Sie die Schritte 1 bis 4 unter Schreiben von Daten
  2. Verwendung sqlContext.read.jdbc:

    sqlContext.read.jdbc(url=url, table="baz", properties=properties)
    

    oder sqlContext.read.format("jdbc"):

    (sqlContext.read.format("jdbc")
        .options(url=url, dbtable="baz", **properties)
        .load())
    

Bekannte Probleme und Fallstricke :

Wo finden Sie geeignete Treiber:

  • Maven Repository (um die erforderlichen Koordinaten für die --packagesAuswahl der gewünschten Version zu erhalten und Daten von einer Gradle-Registerkarte in einem Formular zu kopieren, das die compile-group:name:versionentsprechenden Felder ersetzt) ​​oder Maven Central Repository :

Andere Optionen

Abhängig von der Datenbank kann eine spezialisierte Quelle vorhanden sein, die in einigen Fällen bevorzugt wird:

null323
quelle
Das ist so einfach zu bedienen, unglaublich.
user3072843
mode = "überschreiben" Verwenden Sie diesen Befehl: spark_submit --driver-class-path /xx/yy/postgresql-xx.jar my-script.py
Sanjay Das
-1

Laden Sie den Treiber mysql-connector-java herunter und bewahren Sie ihn im Ordner spark jar auf. Beachten Sie den folgenden Python-Code, indem Sie Daten in "acotr1" schreiben. Wir müssen die Tabellenstruktur acotr1 in der mysql-Datenbank erstellen

    spark = SparkSession.builder.appName("prasadad").master('local').config('spark.driver.extraClassPath','D:\spark-2.1.0-bin-hadoop2.7\jars\mysql-connector-java-5.1.41-bin.jar').getOrCreate()

    sc = spark.sparkContext

    from pyspark.sql import SQLContext

    sqlContext = SQLContext(sc)

    df = sqlContext.read.format("jdbc").options(url="jdbc:mysql://localhost:3306/sakila",driver="com.mysql.jdbc.Driver",dbtable="actor",user="root",password="****").load()

    mysql_url="jdbc:mysql://localhost:3306/sakila?user=root&password=****"

    df.write.jdbc(mysql_url,table="actor1",mode="append")
y durga prasad
quelle
1
Das hat dein Passwort drin! Klicken Sie auf "Bearbeiten" und löschen Sie Ihr Passwort!
ASH
-3

Unter diesem Link können Sie die jdbc für Postgres herunterladen und die Schritte zum Herunterladen der JAR-Datei ausführen

https://jaceklaskowski.gitbooks.io/mastering-apache-spark/exercises/spark-exercise-dataframe-jdbc-postgresql.html JAR-Datei wird in diesem Pfad heruntergeladen. "/home/anand/.ivy2/jars/org.postgresql_postgresql-42.1.1.jar"

Wenn Ihre Funkenversion 2 ist

from pyspark.sql import SparkSession

spark = SparkSession.builder
        .appName("sparkanalysis")
        .config("spark.driver.extraClassPath",
         "/home/anand/.ivy2/jars/org.postgresql_postgresql42.1.1.jar")
        .getOrCreate()

//for localhost database//

pgDF = spark.read \
.format("jdbc") \
.option("url", "jdbc:postgresql:postgres") \
.option("dbtable", "public.user_emp_tab") \
.option("user", "postgres") \
.option("password", "Jonsnow@100") \
.load()


print(pgDF)

pgDF.filter(pgDF["user_id"]>5).show()

Speichern Sie die Datei als Python und führen Sie "python jeweilige Dateiname.py" aus.

anand ml
quelle