Je souhaite lire les données filtrées à partir d'une instance Mysql à l'aide de AWS glue job. Puisqu'une connexion jdbc glue ne me permet pas de pousser vers le bas le prédicat, j'essaie de créer explicitement une connexion jdbc dans mon code.
Je veux exécuter une requête de sélection avec la clause where sur une base de données Mysql en utilisant une connexion jdbc comme indiqué ci-dessous
import com.amazonaws.services.glue.GlueContext
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
object TryMe {
def main(args: Array[String]): Unit = {
val sc: SparkContext = new SparkContext()
val glueContext: GlueContext = new GlueContext(sc)
val spark: SparkSession = glueContext.getSparkSession
// Read data into a DynamicFrame using the Data Catalog metadata
val t = glueContext.read.format("jdbc").option("url","jdbc:mysql://serverIP:port/database").option("user","username").option("password","password").option("dbtable","select * from table1 where 1=1").option("driver","com.mysql.jdbc.Driver").load()
}
}
Cela échoue avec une erreur
com.mysql.jdbc.exceptions.jdbc4.MySQLSyntaxErrorException Vous avez une erreur dans votre syntaxe SQL; consultez le manuel qui correspond à votre Version du serveur MySQL pour la bonne syntaxe à utiliser près de 'select * from table1 où 1 = 1 WHERE 1 = 0 'à la ligne 1
Cela ne devrait-il pas fonctionner? Comment récupérer des données filtrées à l'aide d'une connexion JDBC sans lire la table entière dans un bloc de données?
3 Réponses :
Je pense que le problème est survenu parce que vous n'avez pas utilisé la requête entre parenthèses et fourni un alias. À mon avis, cela devrait ressembler à l'exemple suivant:
val t = glueContext.read.format("jdbc").option("url","jdbc:mysql://serverIP:port/database").option("user","username").option("password","password").option("dbtable","(select * from table1 where 1=1) as t1").option("driver","com.mysql.jdbc.Driver").load()
Plus d'informations sur les paramètres des sources de données SQL:
https://spark.apache.org/docs/latest/sql-data-sources-jdbc.html a>
En ce qui concerne le Glue et le framework fourni par le Glue, il y a aussi l'option "push_down_predicate", mais je n'ai utilisé cette option que sur les sources de données basées sur S3. Je pense que cela ne fonctionne pas sur d'autres sources que sur S3 et les données non partitionnées.
https : //docs.aws.amazon.com/glue/latest/dg/aws-glue-programming-etl-partitions.html
Je vous remercie. Cela a aidé :)
Pourquoi ne pas utiliser create_dynamic_frame_from_options?
Pour tous ceux qui recherchent encore des réponses / exemples supplémentaires, je peux confirmer que l'option push_down_predicate fonctionne avec les sources de données ODBC. Voici comment je lis depuis SQL Server (en Python).
df = glueContext.create_dynamic_frame.from_catalog(database = "db", table_name = "db_dbo_table1", push_down_predicate = "(id >= 2850700 AND statusCode = 'ACT')")
Cela fonctionne aussi mais PAS comme je m'y attendais. Le prédicat n'est pas poussé vers la source de données.
df = glueContext.read.format("jdbc")
.option("url","jdbc:sqlserver://server-ip:port;databaseName=db;")
.option("user","username")
.option("password","password")
.option("dbtable","(select t1.*, t2.name from dbo.table1 t1 join dbo.table2 t2 on t1.id = t2.id) as users")
.option("driver","com.microsoft.sqlserver.jdbc.SQLServerDriver")
.load()
La documentation sur pushDownPredicate indique: L'option permettant d'activer ou de désactiver le déroulement du prédicat dans la source de données JDBC. La valeur par défaut est true , auquel cas Spark poussera les filtres vers la source de données JDBC autant que possible.