Как мне преобразовать список Python в динамический фрейм, который можно использовать для создания CSV-файла в S3 - PullRequest
Новая
       65

Как мне преобразовать список Python в динамический фрейм, который можно использовать для создания CSV-файла в S3

0 голосов
/ 04 апреля 2019

Я использую клей aws с пользовательским скриптом pyspark, который загружает данные из экземпляра aurora и преобразует их. Из-за природы моего источника данных (мне нужно рекурсивно запускать команды sql для списка идентификаторов), я получил обычный список python, который содержит списки кортежей. список выглядит примерно так:
list = [[('id': 1), ('name1': value2),('name2': value2')], [('id': 2),...]

Я попытался преобразовать его в обычный фрейм данных с помощью sparkDateDataFrameMethod: listDataFrame = spark.createDataFrame(list)

и преобразование этого списка в динамический фрейм с использованием метода fromDF класса DynamicFrame, в результате чего получится что-то вроде этого:
listDynamicFrame = fromDF(dataframe = listDataFrame, glue_ctx = glueContext, name = listDynamicFrame)

, а затем передать это методу from_options: 

datasink2 = glueContext.write_dynamic_frame.from_options(frame = 
 listDynamicFrame, connection_type = "s3", connection_options = {"path": "s3://glue.xxx.test"}, 
 format = "csv", transformation_ctx = "datasink2")
job.commit()

Так что да, к сожалению, это не похоже на работу: я получаю следующее сообщение об ошибке: 

1554379793700 final status: FAILED tracking URL: http://169.254.76.1:8088/cluster/app/application_1554379233180_0001 user: root
Exception in thread "main"
org.apache.spark.SparkException: Application application_1554379233180_0001 finished with failed status
at org.apache.spark.deploy.yarn.Client.run(Client.scala:1122)
at org.apache.spark.deploy.yarn.Client$.main(Client.scala:1168)
at org.apache.spark.deploy.yarn.Client.main(Client.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  org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:775)
at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:180) at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:205) at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:119)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
19/04/04 12:10:00 INFO ShutdownHookManager: Shutdown hook called
...