Pyspark - повторное использование соединения JDBC

Question

У меня есть следующая задача:

• загрузка данных из одной таблицы из нескольких схем
• использование PySpark
• использование одного пользователя, который имеет доступ ко всем схемам вDB

Я использую следующий код (более или менее):

def connect_to_oracle_db(spark_session, db_query):
    return spark_session.read \
        .format("jdbc") \
        .option("url", "jdbc:oracle:thin:@//<host>:<port>/<srvice_name") \
        .option("user", "<user>") \
        .option("password", "<pass>") \
        .option("dbtable", db_query) \
        .option("driver", "oracle.jdbc.driver.OracleDriver") 

def run(self):
    all_schemes  = <list of all available schemes>
    for str_schema in all_schemes:
        db_query = "(Select * from " + str_schema + ".TABLE1) TABLE1_DATA"
        df_table1 = slef.connect_to_oracle_db(db_query).load()
        # process df_table1

Существует около 300 схем, и это довольно медленно, потому что каждая для итерации новое соединениесоздан и закрыт.Я хочу найти способ, как повторно использовать существующее соединение или как-то создать пул соединений.Это выглядит довольно неэффективно для меня.

У вас есть идеи, как повторно использовать соединение или создать пул соединений для PySpark?

Answer 1

Нет места для пула соединений в классическом смысле в распределенном, как Spark.Вы должны помнить, что каждый раздел может обрабатываться другим физическим узлом, другим логическим контейнером (если это применимо для данного менеджера кластера) и, наконец, разными JVM.

Не то, чтобы пул соединений мог действительно помочь в таком случае.Поскольку Spark предназначен для массового импорта, использование отдельных подключений уже довольно велико.

Однако здесь есть и другие видимые проблемы (и, возможно, другие проблемы, не очевидные из фрагмента, поскольку код, который вы показали, не 'на самом деле получить данные):

• Вы не настроили fetchsize, поэтому будет использоваться значение по умолчанию для конкретного драйвера.Для Oracle это 10 , что совершенно не подходит для крупномасштабной обработки

  return spark_session.read 
    .format("jdbc")
    .option("fetchsize", some_reasonable_value)
    ...
  

• Вы не настроили разбиение, поэтому Spark обработает вседанные с использованием только одного раздела .О возможных решениях можно прочитать в Как оптимизировать разбиение при переносе данных из источника JDBC?

• Вы смоделировали этот последовательный процесс.Если набор данных каким-либо образом не объединен в нисходящем направлении, было бы лучше представить отдельное задание для каждой таблицы и позволить планировщику оптимизировать вещи в соответствии с доступными ресурсами.

  Вы также можете рассмотреть обработку таблиц параллельно в одном приложении

И еще раз повторюсь - Spark ленив, поэтому основная проблема может быть в каком-то другом месте, а проблемы, перечисленные выше, могутбыть второстепенным.