Erstellen einer aggregierten Metriken aus JSON-Logs in Apache Funke

Question

Ich fange mit Apache Funke an. Ich habe eine Anforderung, ein JSON-Protokoll in eine abgeflachte Metriken zu konvertieren, kann auch als eine einfache CSV betrachtet werden.

Für z.

"orderId":1, 
    "orderData": { 
    "customerId": 123, 
    "orders": [ 
    { 
     "itemCount": 2, 
     "items": [ 
     { 
      "quantity": 1, 
      "price": 315 
     }, 
     { 
      "quantity": 2, 
      "price": 300 
     }, 

     ] 
    } 
    ] 
} 

Dies kann als ein einzelnes json log in Betracht gezogen werden, möchte ich dies in,

orderId,customerId,totalValue,units 
    1 , 123 , 915 , 3 

Ich wollte durch sparkSQL Dokumentation

konvertieren und es halten, einzelne Werte wie „wählen bekommen können orderId, orderData.customerId vom Auftrag "aber ich bin nicht sicher, wie man die Summe aller Preise und Maßeinheiten erhält.

Was sollte die beste Vorgehensweise sein, um dies mit apache spark zu erreichen?

Answer 1

Versuche:

>>> from pyspark.sql.functions import * 
>>> doc = {"orderData": {"orders": [{"items": [{"quantity": 1, "price": 315}, {"quantity": 2, "price": 300}], "itemCount": 2}], "customerId": 123}, "orderId": 1} 
>>> df = sqlContext.read.json(sc.parallelize([doc])) 
>>> df.select("orderId", "orderData.customerId", explode("orderData.orders").alias("order")) \ 
... .withColumn("item", explode("order.items")) \ 
... .groupBy("orderId", "customerId") \ 
... .agg(sum("item.quantity"), sum(col("item.quantity") * col("item.price"))) 

Answer 2

Für die Menschen, die für eine Java-Lösung der oben suchen, benutzen Sie bitte wie folgt vor:

SparkSession spark = SparkSession 
      .builder() 
      .config(conf) 
      .getOrCreate(); 

    SQLContext sqlContext = new SQLContext(spark); 

    Dataset<Row> orders = sqlContext.read().json("order.json"); 
    Dataset<Row> newOrders = orders.select(
      col("orderId"), 
      col("orderData.customerId"), 
      explode(col("orderData.orders")).alias("order")) 
      .withColumn("item",explode(col("order.items"))) 
      .groupBy(col("orderId"),col("customerId")) 
      .agg(sum(col("item.quantity")),sum(col("item.price"))); 
    newOrders.show();