Запуск pyspark для объединения двух фреймов данных и проблема с зависанием pyspark после вывода журналов ...
INFO FileSourceScanExe c: планирование сканирования с упаковкой бинов, максимальный размер: 25933503 байта, стоимость открытия рассматривается как сканирование 4194304 байта.
...
INFO DAGScheduler: поиск вновь запускаемых этапов
INFO DAGScheduler: выполняется: установлено (ShuffleMapStage 6)
Зависание pyspark INFO DAGScheduler: сбой: Set ()
Зависание на длительное время.
Соответствующий фрагмент кода ...
other_table = sparksession.read.format("jdbc") \
.option("url", "jdbc:oracle:thin:@//oracle.domain.local:12345/DBNAME")\
.option("user", DB_CREDS["oracle"]["username"])\
.option("password", DB_CREDS["oracle"]["password"])\
.option("dbtable", f"({PK_SELECTION_QUERY})") \
.option("driver", "oracle.jdbc.driver.OracleDriver") \
.load()
other_table = other_table.dropDuplicates()
print("\n\nfilter table:")
other_table.show(n=3)
orig_order = df.columns # in case PK joining changes original column positions
df = df.join(other_table.select("TX_ID"), "TX_ID", "leftsemi").select(orig_order)
df.show(n=5)
с конфигами. ..
--conf "spark.driver.extraJavaOptions=-XX:+UseG1GC"
--conf "spark.executor.extraJavaOptions=-XX:+UseG1GC"
--conf "spark.sql.autoBroadcastJoinThreshold=-1"
--jars /home/airflow/tmp/ojdbc6.jar
Не знаете, какую дополнительную информацию я могу добавить к этому сообщению, чтобы дать более полезную информацию.
Кто-нибудь с большим опытом работы с pyspark знает, что здесь может происходить? Есть ли еще какие-либо советы по отладке или информация, которая может быть полезна для публикации?