В соответствии с моим вариантом использования я пытаюсь записать данные из s3 в снежинку. Мое приложение написано на Python / Pyspark и работает в кластере AWS EKS.
Код:
def write_from_audience_parquet_to_snowflake(self):
self.logger.info('write_from_audience_parquet_to_snowflake started')
path = os.path.dirname(os.path.abspath(__file__))
parquet_stage = read_template(path + '/sql/staging_audience_attribute.sql')
input_base_s3 = self.base_input_s3_path.replace("s3a://", "s3://")
self.logger.info("input_base_s3 = {}".format(input_base_s3))
parquet_stage = parquet_stage.render(input_base_s3=input_base_s3, mab_id=self.mab_id,
mab_send_id=self.mab_send_id, access_key=self.aws_access_key,
secret_key=self.aws_secret_key, session_token=self.aws_session_key)
create_table_template = read_template(path + '/sql/create_audience_table.sql')
create_table_sql = create_table_template.render(audience_table=self.audience_table)
upload_stage = read_template(path + '/sql/upload_audience_stage.sql')
upload_stage = upload_stage.render(audience_table=self.audience_table)
try:
self.utilSf.runQuery(self.sfoptions, parquet_stage)
self.utilSf.runQuery(self.sfoptions, create_table_sql)
self.utilSf.runQuery(self.sfoptions, upload_stage)
except Exception as e:
self.logger.exception(e)
raise e
self.logger.info('write_from_audience_parquet_to_snowflake completed successfully')
Ошибка:
An error occurred while calling z:net.snowflake.spark.snowflake.Utils.runQuery.
: java.lang.NullPointerException
at java.util.Hashtable.put(Hashtable.java:460)
at net.snowflake.spark.snowflake.JDBCWrapper.getConnector(SnowflakeJDBCWrapper.scala:141)
at net.snowflake.spark.snowflake.Utils$.getJDBCConnection(Utils.scala:244)
at net.snowflake.spark.snowflake.Utils$.getJDBCConnection(Utils.scala:249)
at net.snowflake.spark.snowflake.Utils$.runQuery(Utils.scala:391)
at net.snowflake.spark.snowflake.Utils.runQuery(Utils.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.lang.Thread.run(Thread.java:748)
Пожалуйста, найдите полный журнал ошибок здесь
Я прочитал, что это происходит из-за проблемы с версионированием scala, которая возникает в spark. Я использую спарк 2.4.5 с внешним добавлением oop 3.1.2. Также я использую следующие баночки снежинок:
spark-snowke_2.11-2.4.10-spark_2.4.jar
Snowkeke-Ingest-SDK-0.9.5.jar
Snowkeke-JDB c -3.6.15.jar
Если вам нужна дополнительная информация, пожалуйста, сообщите мне об этом в комментариях.