Apache Spark 运算符¶
前提条件¶
要使用
SparkSubmitOperator
,您必须配置 Spark 连接。要使用
SparkJDBCOperator
,您必须同时配置 Spark 连接 和 JDBC 连接。SparkSqlOperator
的所有配置均来自运算符参数。
SparkJDBCOperator¶
在 Apache Spark 服务器上启动应用程序,它使用 SparkSubmitOperator
在基于 JDBC 的数据库之间进行数据传输。
有关参数定义,请参阅 SparkJDBCOperator
。
使用运算符¶
使用 cmd_type
参数,可以在 Spark 与数据库之间传输数据(spark_to_jdbc
),或在数据库与 Spark 之间传输数据(jdbc_to_spark
),后者将使用 Spark 命令 saveAsTable
写入表。
tests/system/apache/spark/example_spark_dag.py
jdbc_to_spark_job = SparkJDBCOperator(
cmd_type="jdbc_to_spark",
jdbc_table="foo",
spark_jars="${SPARK_HOME}/jars/postgresql-42.2.12.jar",
jdbc_driver="org.postgresql.Driver",
metastore_table="bar",
save_mode="overwrite",
save_format="JSON",
task_id="jdbc_to_spark_job",
)
spark_to_jdbc_job = SparkJDBCOperator(
cmd_type="spark_to_jdbc",
jdbc_table="foo",
spark_jars="${SPARK_HOME}/jars/postgresql-42.2.12.jar",
jdbc_driver="org.postgresql.Driver",
metastore_table="bar",
save_mode="append",
task_id="spark_to_jdbc_job",
)
参考¶
有关更多信息,请参阅 Apache Spark DataFrameWriter 文档。
SparkSqlOperator¶
在 Apache Spark 服务器上启动应用程序,这要求 spark-sql
脚本位于 PATH 中。该运算符将在 Spark Hive 元存储服务上运行 SQL 查询,sql
参数可以模板化,可以是 .sql
或 .hql
文件。
有关参数定义,请参阅 SparkSqlOperator
。
使用运算符¶
tests/system/apache/spark/example_spark_dag.py
spark_sql_job = SparkSqlOperator(
sql="SELECT COUNT(1) as cnt FROM temp_table", master="local", task_id="spark_sql_job"
)
参考¶
有关更多信息,请参阅 运行 Spark SQL CLI。
SparkSubmitOperator¶
在 Apache Spark 服务器上启动应用程序,它使用 spark-submit
脚本来设置 Spark 及其依赖项的类路径,并支持 Spark 支持的不同集群管理器和部署模式。
有关参数定义,请参阅 SparkSubmitOperator
。
使用运算符¶
tests/system/apache/spark/example_spark_dag.py
submit_job = SparkSubmitOperator(
application="${SPARK_HOME}/examples/src/main/python/pi.py", task_id="submit_job"
)
参考¶
有关更多信息,请参阅 Apache Spark 提交应用程序。