This pull request has centrally tracked the following features:

(1) create initial version of entroy UDFs issue: https://github.com/linkedin/datafu/issues/77 (2) experiment and implement weighted sampling algorithms issue: https://github.com/linkedin/datafu/issues/80

Hi, I've wrote UDF in groovy using AliasableEvalFunc

I want to test it on unit level, I don't want to wrap this class with pig script. How can I push down a tuple schema to AliasableEvalFunc subcalss? I can't find example in tests.

I've wrote such ugly code, looks like it works: ``groovy

ZonesGenerator initUdf(){ def getAliasSchemaMap = { def aliasSchemaAsStr = 'bag1.field1=0, bag2::moreName.field1=1' aliasSchemaAsStr.split(',').inject([:]){memo, keyVal -> memo[keyVal.split('=').first()] = Integer.valueOf(keyVal.split('=').last()) return memo } }

    new ZonesGenerator(){
        public Map<String, Integer> getFieldAliases(){
            return getAliasSchemaMap
        }
    }

}

``

In the real world, there are occasions we need to calculate the entropy of discrete random variables, for instance, to calculate the mutual information between variable X and Y using its entropy-based formula(mutual information calculation could be found at http://en.wikipedia.org/wiki/Mutual_information#Relation_to_other_quantities). Would suggest to implement a UDF to calculate the entropy of given input samples, following the definition at http://en.wikipedia.org/wiki/Entropy_%28information_theory%29

In Apache Mahout, there is an existing implementation of entropy using Map/Reduce API, please refer to https://issues.apache.org/jira/browse/MAHOUT-747.

UDF Spec:

package datafu.pig.stats;

public class Entropy extends AccumulatorEvalFunc

Constructor accepts a string to indicate its logarithm base, including: 2, e(Euler's number), and 10

Input: a bag of ordered tuples X = {x[0], x[1], ... x[N - 1]}

Output: entropy value of double type

How to use:

X = LOAD 'input' AS (val:int);
EX = FOREACH (GROUP X ALL) {OX = ORDER X by val; GENERATE Entropy(OX);}

This is the reference paper I use to learn about the weighted sampleing algorithm: http://utopia.duth.gr/~pefraimi/research/data/2007EncOfAlg.pdf

The present WeightedSample.java implements the Algorithm D.

We may try Algorithm A, A-res and A-expJ since they could be used in a data stream and distributed environment. These algorithms could be implemented based on ReservoirSample.java(inherit from this class?) since they also need a reservior to store the selected items.

I've moved the hard coded target version to a property, so that the default behavior will remain the same. However user will have ability to override the target version via the property on command line, for example:

ant clean jar -DtargetJavaVersion=1.7

file ./topics.pig, line 31, column 62> Failed to generate logical plan. Nested exception: java.lang.RuntimeException: could not instantiate 'datafu.pig.stats.Quantile' with arguments 'null'

When:

quantiles = foreach (group token_counts all) generate FLATTEN(datafu.pig.stats.Quantile('0.10', '0.90')) as (low_ten, high_ten);

HI when using ReservoirSample it seems like the sample is done on the full input instead of the group input.

e.g. Lets say my input.txt is a1,5 a1,6 a1,7 a2,5 a2,6 a2,7

I have the following program: DEFINE SRS datafu.pig.sampling.ReservoirSample('2'); data = LOAD 'input.txt' USING PigStorage(',') AS (key: chararray, value: chararray); grouped = GROUP data BY key; sample2 = FOREACH grouped GENERATE FLATTEN(SRS(data));

The output was. I would assume that that for each group it would sample (meaning I would see 2 samples of a1 and 2 samples of a2...why is that not)? (a1,7) (a1,6) (a1,6) (a1,6)

Pig 0.12.0 has added several new keywords such as "Assert" or "In" (via PIG-3367 and PIG-3269). This is causing issues to DataFu UDFs with the same name. I wasn't able to find any other solution other then renaming the UDFs in question. I do understand that this is very intrusive change, so I've provided deprecated ones for backward compatibility - e.g users on Pig 0.11.0 will still be able to use the Assert or In as they are now.

After applying this final set of changes I was able to run all unit tests against Pig 0.12.0 without any issues!

The StreamingQuantile UDF generates intermediary quantiles but outputs only the request quantiles. However, the outputSchema method does not reflect that. So the following Pig script breaks with an incompatible schema error:

define Quantiles datafu.pig.stats.StreamingQuantile('0.5', '0.9', '0.95', '1'); A = foreach B generate flatten(Quantiles(B.C)) as (q1, q2, q3, q4);

I changed the outputSchema so that only the requested quantiles are in the schema.

Lead is an analytic function like Oracle's Lead function. It provides access to more than one tuple of a bag at the same time without a self join. Given a bag of tuple returned from a query, LEAD provides access to a tuple at a given physical offset beyond that position. Generates pairs of all items in a bag.

If you do not specify offset, then its default is 1. Null is returned if the offset goes beyond the scope of the bag.

Example 1:

   register ba-pig-0.1.jar

   define Lead datafu.pig.bags.Lead('2');

   -- INPUT: ({(1),(2),(3),(4)})
   data = LOAD 'input' AS (data: bag {T: tuple(v:INT)});
   describe data;

   -- OUTPUT:  ({((1),(2),(3)),((2),(3),(4)),((3),(4),),((4),,)})
   -- OUTPUT SCHEMA: data2: {lead_data: {(elem0: (v: int),elem1: (v: int),elem2: (v: int))}}
   data2 = FOREACH data GENERATE Lead(data);
   describe data2;
   DUMP data2;

Example 2

   register  ba-pig-0.1.jar

   define Lead datafu.pig.bags.Lead();

   -- INPUT: ({(10,{(1),(2),(3)}),(20,{(4),(5),(6)}),(30,{(7),(8)}),(40,{(9),(10),(11)}),(50,{(12),(13),(14),(15)})})
   data = LOAD 'input' AS (data: bag {T: tuple(v1:INT,B: bag{T: tuple(v2:INT)})});
   --describe data;

   -- OUPUT: ({((10,{(1),(2),(3)}),(20,{(4),(5),(6)})),((20,{(4),(5),(6)}),(30,{(7),(8)})),((30,{(7),(8)}),(40,{(9),(10),(11)})),((40,{(9),(10),(11)}),(50,{(12),(13),(14),(15)})),((50,{(12),(13),(14),(15)}),)})
   data2 = FOREACH data GENERATE Lead(data);
   --describe data2;
   DUMP data2;

This UDF takes an input bag which is assumed to be ordered by an ISO8601 timestamp field and returns the total time elapsed (in seconds) by all tuples in the bag. The first field in each of the tuples in the DataBag is assumed to be the ISO8601 timestamp.

Example:

 DEFINE TotalTimeBetween datafu.pig.bags.TotalTimeBetween();

-- input: 
-- 2012-01-01T00:00:00\tbiuA8n98wn\thttp://www.google.com/
-- 2012-01-01T00:00:00\tasd909m09j\thttp://www.google.com/
-- 2012-01-01T00:01:00\tbiuA8n98wn\thttp://www.google.com/1
-- 2012-01-01T00:01:05\tbiuA8n98wn\thttp://www.google.com/
-- 2012-01-01T00:02:00\tasd909m09j\thttp://www.google.com/2
input = LOAD 'input' AS (timestamp:chararray, visitor_id:chararray, url:chararray);
by_visitor = GROUP data BY visitor_id;
count_time = FOREACH by_visitor {
    ordered = ORDER data BY timestamp ASC;
    GENERATE group, TotalTimeBetween(ordered);
};
DUMP count_time;
-- (biuA8n98wn,65)
-- (asd909m09j,120)

In truth I wasn't really sure what to call / where to put this UDF so I'm open to suggestions.

Thanks for the fix to SampleByKey issue. Please let us know when can we expect the release that contains this fix.

Or If the build instructions are documented somewhere I can get the appropriate patch and get a temporary build till the official JAR is out.

Thanks Bala

I have a pair of 35M of links from 117K nodes and ran pagerank job on 3 node m2.2xlarge EMR cluster. Initially I got out of memory error in the reduce phase so I increased the JVM size and then now I am getting the following error (and this happens in one reduce job and the other 3 completes without any error):

2015-01-04 03:44:40,349 WARN [main] org.apache.hadoop.mapred.YarnChild: Exception running child : org.apache.pig.backend.executionengine.ExecException: ERROR 0: Exception while executing (Name: topic_ranks: New For Each(false,false,false)[bag] - scope-42 Operator Key: scope-42): org.apache.pig.backend.executionengine.ExecException: ERROR 0: Exception while executing [POUserFunc (Name: POUserFunc(datafu.pig.linkanalysis.PageRank)[bag] - scope-33 Operator Key: scope-33) children: null at []]: java.lang.NullPointerException at org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator.processInput(PhysicalOperator.java:289) at org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POForEach.getNextTuple(POForEach.java:242) at org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigGenericMapReduce$Reduce.runPipeline(PigGenericMapReduce.java:464) at org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigGenericMapReduce$Reduce.processOnePackageOutput(PigGenericMapReduce.java:432) at org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigGenericMapReduce$Reduce.reduce(PigGenericMapReduce.java:412) at org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigGenericMapReduce$Reduce.reduce(PigGenericMapReduce.java:256) at org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:171) at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:635) at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:390) at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:167) at java.security.AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:415) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1548) at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:162)

Any idea how to resolve this issue? I used Hadoop 2.4.0 and Pig 0.12.0.

A couple of different configuration options are available for calculating ndcg. You can specify positional values per range, use a standard logarithmic discounting function, or use a custom function.