Calculer les différences entre successives des enregistrements dans Hadoop avec des Requêtes Hive

C'est une vieille question, mais pour de futures références, j'écris ici une autre proposition:

Ruche Les fonctions de fenêtrage permet d'utiliser les boutons précédent /suivant les valeurs de votre requête.

Un simili code de requête peut être :

SÉLECTIONNEZ customer_id, GAL(call_time, 1, 0) OVER (PARTITION BY customer_id COMMANDE PAR call_time LIGNES 1 ci-dessus) - call_time from matable;

OriginalL'auteur jbaptiste

Peut-être quelqu'un rencontre une exigence similaire, la solution que j'ai trouvée est la suivante:

1) Créer une fonction personnalisée:

package com.example;
//imports (they depend on the hive version)
@Description(name = "delta", value = "_FUNC_(customer id column, call time column) "
    + "- computes the time passed between two succesive records from the same customer. "
    + "It generates 3 columns: first contains the customer id, second contains call time "
    + "and third contains the time passed from the previous call. This function returns only "
    + "the records that have a previous call from the same customer (requirements are not applicable "
    + "to the first call)", extended = "Example:\n> SELECT _FUNC_(customer_id, call_time) AS"
    + "(customer_id, call_time, time_passed) FROM (SELECT customer_id, call_time FROM mytable "
    + "DISTRIBUTE BY customer_id SORT BY customer_id, call_time) t;")
public class DeltaComputerUDTF extends GenericUDTF {
private static final int NUM_COLS = 3;

private Text[] retCols; //array of returned column values
private ObjectInspector[] inputOIs; //input ObjectInspectors
private String prevCustomerId;
private Long prevCallTime;

@Override
public StructObjectInspector initialize(ObjectInspector[] ois) throws UDFArgumentException {
    if (ois.length != 2) {
        throw new UDFArgumentException(
                "There must be 2 arguments: customer Id column name and call time column name");
    }

    inputOIs = ois;

    //construct the output column data holders
    retCols = new Text[NUM_COLS];
    for (int i = 0; i < NUM_COLS; ++i) {
        retCols[i] = new Text();
    }

    //construct output object inspector
    List<String> fieldNames = new ArrayList<String>(NUM_COLS);
    List<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>(NUM_COLS);
    for (int i = 0; i < NUM_COLS; ++i) {
        //column name can be anything since it will be named by UDTF as clause
        fieldNames.add("c" + i);
        //all returned type will be Text
        fieldOIs.add(PrimitiveObjectInspectorFactory.writableStringObjectInspector);
    }

    return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);
}

@Override
public void process(Object[] args) throws HiveException {
    String customerId = ((StringObjectInspector) inputOIs[0]).getPrimitiveJavaObject(args[0]);
    Long callTime = ((LongObjectInspector) inputOIs[1]).get(args[1]);

    if (customerId.equals(prevCustomerId)) {
        retCols[0].set(customerId);
        retCols[1].set(callTime.toString());
        retCols[2].set(new Long(callTime - prevCallTime).toString());
        forward(retCols);
    }

    //Store the current customer data, for the next line
    prevCustomerId = customerId;
    prevCallTime = callTime;
}

@Override
public void close() throws HiveException {
    //TODO Auto-generated method stub

}

}

2) Créer un jar contenant cette fonction. Supposons que le jarname est myjar.jar.

3) Copier le jar de la machine avec de la Ruche. Supposons qu'il est placé dans /tmp

4) Définir la fonction personnalisée à l'intérieur de la Ruche:

ADD JAR /tmp/myjar.jar;
CREATE TEMPORARY FUNCTION delta AS 'com.example.DeltaComputerUDTF';

5) l'Exécution de la requête:

SELECT delta(customer_id, call_time) AS (customer_id, call_time, time_difference) FROM 
  (SELECT customer_id, call_time FROM mytable DISTRIBUTE BY customer_id SORT BY customer_id, call_time) t;

Remarques:

un. J'ai supposé que le call_time colonne stocke les données comme bigint. Dans le cas où c'est une chaîne, dans les processus de la fonction nous récupérer en tant que chaîne de caractères (comme nous le faisons avec le code client), puis de l'analyser à Long

b. J'ai décidé d'utiliser un UDTF au lieu de l'UDF, car de cette façon, il génère toutes les données dont il a besoin. Dans le cas contraire (UDF) les données générées doit être filtrée pour ignorer les valeurs NULL. Ainsi, avec l'UDF fonction (DeltaComputerUDF) décrit dans la première édition de l'original post, la requête sera:

SELECT customer_id, call_time, time_difference 
FROM 
  (
    SELECT delta(customer_id, call_time) AS (customer_id, call_time, time_difference) 
    FROM 
      (
         SELECT customer_id, call_time FROM mytable
         DISTRIBUTE BY customer_id
         SORT BY customer_id, call_time
       ) t
   ) u 
WHERE time_difference IS NOT NULL;

c. Les deux fonctions (UDF et UDTF) comme on le souhaite, peu importe l'ordre des lignes à l'intérieur de la table (donc il n'est pas nécessaire que les données de la table, triées par numéro de client et le temps de garde avant d'utiliser les fonctions delta)

OriginalL'auteur Cipi