Как преобразовать результат JSON в паркет? - PullRequest
Новая
       41

Как преобразовать результат JSON в паркет?

0 голосов
/ 12 июня 2018

У меня есть следующий код, который получает некоторые данные из системы Marketo 

from marketorestpython.client import MarketoClient
munchkin_id = "xxx-xxx-xxx"
client_id = "00000000-0000-0000-0000-00000000000"
client_secret= "secret"
mc = MarketoClient(munchkin_id, client_id, client_secret)
mc.execute(method='get_multiple_leads_by_filter_type', filterType='email', filterValues=['email@domain.com'], 
                  fields=['BG__c','email','company','createdAt'], batchSize=None)

Это возвращает мне следующие данные 

[{'BG__c': 'ABC',
  'company': 'MCS',
  'createdAt': '2016-10-25T14:04:15Z',
  'id': 4,
  'email': 'email@domain.com'},
 {'BG__c': 'CDE',
  'company': 'MSC',
  'createdAt': '2018-03-28T16:41:06Z',
  'id': 10850879,
  'email': 'email@domain.com'}]

Что я хочу сделать, чтобы сохранить этовернулся в файл паркета.Но когда я пытаюсь сделать это с помощью следующего кода, я получаю сообщение об ошибке. 

from marketorestpython.client import MarketoClient
munchkin_id = "xxx-xxx-xxx"
client_id = "00000000-0000-0000-0000-00000000000"
client_secret= "secret"
mc = MarketoClient(munchkin_id, client_id, client_secret)
data = mc.execute(method='get_multiple_leads_by_filter_type', filterType='email', filterValues=['email@domain.com'], 
                  fields=['BG__c','email','company','createdAt'], batchSize=None)

sqlContext.read.json(data)
data.write.parquet("adl://subscription.azuredatalakestore.net/folder1/Marketo/marketo_data")

java.lang.ClassCastException: java.util.HashMap cannot be cast to java.lang.String
---------------------------------------------------------------------------
Py4JJavaError                             Traceback (most recent call last)
<command-1431708582476650> in <module>()
      7                       fields=['BG__c','email','company','createdAt'], batchSize=None)
      8 
----> 9 sqlContext.read.json(data)
     10 data.write.parquet("adl://subscription.azuredatalakestore.net/folder1/Marketo/marketo_data")

/databricks/spark/python/pyspark/sql/readwriter.py in json(self, path, schema, primitivesAsString, prefersDecimal, allowComments, allowUnquotedFieldNames, allowSingleQuotes, allowNumericLeadingZero, allowBackslashEscapingAnyCharacter, mode, columnNameOfCorruptRecord, dateFormat, timestampFormat, multiLine, allowUnquotedControlChars, charset)
    261             path = [path]
    262         if type(path) == list:
--> 263             return self._df(self._jreader.json(self._spark._sc._jvm.PythonUtils.toSeq(path)))
    264         elif isinstance(path, RDD):
    265             def func(iterator):

/databricks/spark/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py in __call__(self, *args)
   1158         answer = self.gateway_client.send_command(command)
   1159         return_value = get_return_value(
-> 1160             answer, self.gateway_client, self.target_id, self.name)
   1161

Что я делаю не так?

Ответы [ 2 ]

0 голосов
/ 12 июня 2018

В соответствии с приведенным ниже утверждением в вашем коде вы непосредственно пишете данные.Вы должны сначала создать фрейм данных.Вы можете конвертировать json в df, используя val df = sqlContext.read.json ("путь / к / json / file"). Затем выполните df.write 

data.write.parquet("adl://subscription.azuredatalakestore.net/folder1/Marketo/marketo_data")
0 голосов
/ 12 июня 2018

У вас есть следующие данные 

data = [{'BG__c': 'ABC',
       'company': 'MCS',
       'createdAt': '2016-10-25T14:04:15Z',
       'id': 4,
       'email': 'email@domain.com'},
       {'BG__c': 'CDE',
       'company': 'MSC',
       'createdAt': '2018-03-28T16:41:06Z',
       'id': 10850879,
       'email': 'email@domain.com'}]

Чтобы сохранить их в файле паркета, я бы предложил создать DataFrame, чтобы затем сохранить его как паркет. 

from pyspark.sql.types import *

df = spark.createDataFrame(data,
                           schema = StructType([
                                    StructField("BC_g", StringType(), True),
                                    StructField("company", StringType(), True),
                                    StructField("createdAt", StringType(), True),
                                    StructField("email", StringType(), True),
                                    StructField("id", IntegerType(), True)]))

Это даст следующие типы: 

df.dtypes

[('BC_g', 'string'),
 ('company', 'string'),
 ('createdAt', 'string'),
 ('email', 'string'),
 ('id', 'int')]

Затем вы можете сохранить фрейм данных в виде файла паркета 

df.show()
+-----+-------+--------------------+----------------+--------+
|BG__c|company|           createdAt|           email|      id|
+-----+-------+--------------------+----------------+--------+
|  ABC|    MCS|2016-10-25T14:04:15Z|email@domain.com|       4|
|  CDE|    MSC|2018-03-28T16:41:06Z|email@domain.com|10850879|
+-----+-------+--------------------+----------------+--------+

df.write.format('parquet').save(parquet_path_in_hdfs)

Где parquet_path_in_hdfs - это путь и имя нужного файла паркета

...