AzureデータレイクストアでJSONをCSVに変換するPythonを使用したU-SQL

Question

Azureデータレイクストアに格納されている巨大なファイルをネストJSONからCSVに変換する必要があります。 Pythonモジュールpandas、numpyはAzureデータレイク解析でサポートされているので、標準モジュールとは別にPythonを使用してこれを実現することは可能です。誰もこれを達成するためのPythonコードを持っていますか？

ソース形式：

{ "のLoc"： "TDM"、 "トピック"： "位置"、 "LocMac"：「位置/ FE：7A：XX：XX：XX：XX "、" seq "：" 296083773 "、"タイムスタンプ "：1488986751、" op "：" OP_UPDATE "、" topicSeq "：" 46478211 "、" sourceId "：" AFBWmHSe "、" location "：{" staEthMac " "addr"： "/ xxxxx"}、 "staLocationX"：1643.8915、 "staLocationY"：571.04205、 "errorLevel"：1076、 "associated"：0、 "campusId"： "n5THo6IINuOSVZ/cTidNVA =="、 "buildingId" "ALETM"、 "unit"： "FEET"}、 "EventProcessedUtcTime"は、 "7hY/xx =="、 "floorId"： "xxxxxxxxxx + BYoo0A =="、 "hashedStaEthMac"： "xxxx/pMVyK4Gu9qG6w ="、 "locAlgorithm"： "ALGORITHM_ESTIMATION" "：" 2017-03-08T15：35：02.3847947Z "、" PartitionId "：3、" EventEnqueuedUtcTime "：" 2017-03-08T15：35：03.7510000Z "、" IoTHub "：{" Mess 「EnqueuedTime」：「0001-01-01T00：00：00.0000000」、「StreamId」：NULL}}：null、「CorrelationId」：null、「ConnectionDeviceId」：「xxxxx」、「ConnectionDeviceGenerationId」：「636243184116591838」、「EnqueuedTime」

期待出力

TDM、位置、場所/ 80：7A：BF：D4：D6：50,974851970,1490004475、OP_UPDATE、151002334、XXXXXXX、GHQ/1NZQ、977.7259、 638.8827,490,1、n5THo6IINuOSVZ/cTidNVA ==、7hY/jVh9NRqqxF6gbqT7Jw ==、LV/ZiQRQMS2wwKiKTvYNBQ ==、H5rrAD/jg1Fnkmo1Zmquau/Qn1U =、ALGORITHM_ESTIMATION、FEET

Answer 1

あなたの説明によると、私の理解に基づいて、Azure Data Lake Storeに保存されているデータをPythonでJSON形式からCSV形式に変換する方法があると思います（pandas/numpyパッケージ）。そこでソースデータを見て、JSONに配列型がないと仮定して、サンプルデータ変換のために以下のコードを設計しました。

JSON形式のオブジェクト文字列のサンプルコードです。参考として、私は自分の考えを理解するために、flattern構造体{"A": 0, "B": {"C": 1}}を構造体[["A", "B.C"], [0, 1]]に変換する方法を理解するためのコメントを追加します。

import json 
import pandas as pd 

# Source Data string 
json_raw = '''{"Loc":"TDM","Topic":"location","LocMac":"location/fe:7a:xx:xx:xx:xx","seq":"296083773","timestamp":1488986751,"op":"OP_UPDATE","topicSeq":"46478211","sourceId":"AFBWmHSe","location":{"staEthMac":{"addr":"/xxxxx"},"staLocationX":1643.8915,"staLocationY":571.04205,"errorLevel":1076,"associated":0,"campusId":"n5THo6IINuOSVZ/cTidNVA==","buildingId":"7hY/xx==","floorId":"xxxxxxxxxx+BYoo0A==","hashedStaEthMac":"xxxx/pMVyK4Gu9qG6w=","locAlgorithm":"ALGORITHM_ESTIMATION","unit":"FEET"},"EventProcessedUtcTime":"2017-03-08T15:35:02.3847947Z","PartitionId":3,"EventEnqueuedUtcTime":"2017-03-08T15:35:03.7510000Z","IoTHub":{"MessageId":null,"CorrelationId":null,"ConnectionDeviceId":"xxxxx","ConnectionDeviceGenerationId":"636243184116591838","EnqueuedTime":"0001-01-01T00:00:00.0000000","StreamId":null}}''' 

# Load source data string to a Python dict 
json_data = json.loads(json_raw) 

# The key method `flattern` for converting `dict` to `2D-list` 
def flattern(data, key): 
    keys = [] 
    values = [] 
    if key is None: 
     for key in data: 
      if type(data[key]) is dict: 
       keys.extend(flattern(data[key], key)[0]) 
       values.extend(flattern(data[key], key)[1]) 
      else: 
       keys.append(key) 
       values.append(data[key]) 
    else: 
     for subkey in data: 
      if type(data[subkey]) is dict: 
       keys.extend(flattern(data[subkey], key+"."+subkey)[0]) 
       values.extend(flattern(data[subkey], subkey)[1]) 
      else: 
       keys.append(key+"."+subkey) 
       values.append(data[subkey]) 
    return [keys, values] 

list2D = flattern(json_data, None) 
df = pd.DataFrame([list2D[1],], columns=list2D[0]) 

# If you want to extract the items `Loc` & `Topic` & others like `location.staEthMac.addr`, you just need to create a list for them. 
selected = ["Loc", "Topic"] 
# Use `selected` list to select the columns you want. 
result = df.ix[:,selected] 
# Transform DataFrame to csv string 
csv_raw = "\n".join([",".join(lst) for lst in pd.np.array(result)]) 

希望します。