Выполнение сценариев Spark Scala внутри запущенного приложения Spark. ⇐ JAVA

[Anonymous]

У меня есть приложение Java, которое выполняет некоторые операции конвейера Spark, например:

Код: Выделить всё

public class SparkJobExample {

public void run(...) {

var spark = SparkSession.builder()
.master("local[*]")
.getOrCreate();

//Load initial df
var initDf = spark.read()
.csv(
);
initDf.createOrReplaceTempView("initDf");

//Operation 1
firstDf = 
firstDf.createOrReplaceTempView("firstDf");

...

//Operation N
nDf = 
nDf.createOrReplaceTempView("nDf");

//Save result df...
}
}

Он работает очень хорошо, но я хотел бы динамически запускать некоторые операции из Spark Script Scala, представленного в виде файла, например:

Код: Выделить всё

import org.apache.spark.sql.SparkSession

object DynamicSparkScript {
def main(args: Array[String]): Unit = {

// Get the same session which was initialized in regular java code
val spark = SparkSession.builder().getOrCreate()

// Read DataFrame which was created in regular java code and saved as temp view
val df = spark.table()

val resultDf = df... //some operations on df read above

resultDf.createOrReplaceTempView("scriptRes")
}

Например, после выполнения операции N в обычном Java-коде я хотел бы выполнить этот сценарий Spark Scala, который использует View, сохраненный в Java-коде, выполняет некоторые операции и сохраняет результат DF, который позже доступен из кода Java:

Код: Выделить всё

        //Java Code

//Operation N
nDf = 
nDf.createOrReplaceTempView("nDf");

//Read and execute scala script which creates View name "scriptRes"
var scalaScriptPath =



//Use DF created in script
var fromScriptDf = spark.table("scriptRes")
finalDf = fromScriptDf. ...

//save final df

Если я правильно понимаю, нет проблем с передачей DF таким образом, поскольку он должен быть доступен в Spark Session.
Однако я не могу понять, как выполнить и правильно запустить этот скрипт Scala. Я получаю путь к этому файлу динамически, когда приложение уже запущено, оно недоступно при запуске. Обратите внимание, что я не могу использовать объект SparkLauncher внутри обычного кода.
У меня было две идеи:

• Упаковка этого сценария scala в jar и запуск его как другого процесса из кода Java. Кажется, это может сработать, хотя для этого мне потребуется упаковать все библиотеки Spark в этот jar-файл, что сделает его тяжелым, и все действие, вероятно, будет очень медленным (особенно если будет несколько разных сценариев)
• Использование динамической загрузки классов и API отражения Java для создания этого класса во время выполнения и вызова его метода.

Есть ли какие-либо специальные функции или лучше способ сделать это? Я думал, можно ли просто прочитать код из файла и каким-то образом передать его в существующий SparkSession, но ничего не нашел.
Если нет, то какое решение, по вашему мнению, лучше?

Подробнее здесь: https://stackoverflow.com/questions/784 ... pplication