基于Spark版本的火花负载罐

Question

我们希望以版本不可知的方式为我们的spark作业加载< code > org . Apache . spark:spark-avro _ 2.12 包。

有的人在用火花3.1.2，有的人在用3.2.0。

对于3.1.2上的那些，我需要加载：

org.apache.spark:spark-avro_2.12:3.1.2

对于3.2.0上的，我需要加载：

org.apache.spark:spark-avro_2.12:3.2.0

我是否可以实现一种通用机制来根据火花版本加载正确的jar，而无需我的用户在他们那端做任何事情？

我不介意在本地保存所有版本的jar，并通过--jars或spark.jars添加它们。问题是根据用户的火花版本选择正确的jar。

对于完整的上下文，我想将其配置为与< code>pytest-spark一起使用。

Answer 1

这是我在< code>pytest-spark上下文中想到的，尽管应该是普遍适用的:

@pytest.fixture(scope='session')
def spark_session(request, spark_conf):
    from pyspark.sql import SparkSession
    import pytest_spark

    spark_conf = pytest_spark.SparkConfigBuilder().get()
    spark_home = pytest_spark.SparkHome(pytest_config=request.config)
    spark_version = re.findall('^Spark (\d+\.\d+\.\d+)', spark_home.version)[0]

    jars = spark_conf.get('spark.jars.packages')
    jars = set(jars.split(',')) if jars else set()
    jars.add(f'org.apache.spark:spark-avro_2.12:{spark_version}')
    spark_conf.set('spark.jars.packages', ",".join(list(jars)))

    session = SparkSession.builder \
        .config(conf=spark_conf) \
        .getOrCreate()

    logger = session.sparkContext._jvm.org.apache.log4j
    logger.LogManager.getLogger("org").setLevel(logger.Level.OFF)
    logger.LogManager.getLogger("akka").setLevel(logger.Level.OFF)

    yield session
    session.stop()