地球人你好!
我正在使用 Airflow 来安排和运行 Spark 任务。
这次我发现的只是 Airflow 可以管理的 python DAG。
DAG 示例:
spark_count_lines.py
import logging
from airflow import DAG
from airflow.operators import PythonOperator
from datetime import datetime
args = {
'owner': 'airflow'
, 'start_date': datetime(2016, 4, 17)
, 'provide_context': True
}
dag = DAG(
'spark_count_lines'
, start_date = datetime(2016, 4, 17)
, schedule_interval = '@hourly'
, default_args = args
)
def run_spark(**kwargs):
import pyspark
sc = pyspark.SparkContext()
df = sc.textFile('file:///opt/spark/current/examples/src/main/resources/people.txt')
logging.info('Number of lines in people.txt = {0}'.format(df.count()))
sc.stop()
t_main = PythonOperator(
task_id = 'call_spark'
, dag = dag
, python_callable = run_spark
)
问题是我不擅长 Python 代码并且有一些用 Java 编写的任务。我的问题是如何在 python DAG 中运行 Spark Java jar?或者也许还有其他方法可以做到这一点?我发现 spark 提交:http://spark.apache.org/docs/latest/submitting-applications.html
但我不知道如何将所有东西连接在一起。也许有人以前使用过它并且有工作示例。感谢您的宝贵时间!
最佳答案
您应该能够使用 BashOperator
。保持其余代码不变,导入所需的类和系统包:
from airflow.operators.bash_operator import BashOperator
import os
import sys
设置所需路径:
os.environ['SPARK_HOME'] = '/path/to/spark/root'
sys.path.append(os.path.join(os.environ['SPARK_HOME'], 'bin'))
并添加运算符:
spark_task = BashOperator(
task_id='spark_java',
bash_command='spark-submit --class {{ params.class }} {{ params.jar }}',
params={'class': 'MainClassName', 'jar': '/path/to/your.jar'},
dag=dag
)
您可以使用 Jinja 模板轻松扩展它以提供额外的参数。
您当然可以通过将 bash_command
替换为适合您情况的模板来针对非 Spark 场景进行调整,例如:
bash_command = 'java -jar {{ params.jar }}'
并调整params
。
关于java - 如何在 Airflow 中运行 Spark 代码?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/39827804/