CODEWITHSUNDEEP
apache-pig
suerte
suerte

Reputation: 73

PIG FILTER relation with next row the same relation

i'm searching for a long time now to solve my problem but nearly found nothing helpful. Hopefully some of you can give me a tip.

I have a relation A with the following format: username, timestamp, ip

For example:

Harald 2014-02-18T16:14:49.503Z 123.123.123.123
Harald 2014-02-18T16:14:51.503Z 123.123.123.123
Harald 2014-02-18T16:14:55.503Z 321.321.321.321

And i want to find out, who changed his ip adress in less then 5 seconds. So the second and the third row should be interesting.

I want do group the relation by username und want to compare the timestamp of the actuall row with the next row. if the ip adress isnt the same and the timestamp is less then 5 seconds bigger, this should be at the output.

could someone help me with that issue?

regards.

first i want to thank you for your time.

but i actually stuck at the Sessionize part.

this is my data comming in:

aoebcu  2014-02-19T14:23:17.503Z    220.61.65.25
aoebcu  2014-02-19T14:23:14.503Z    222.117.144.19
aoebcu  2014-02-19T14:23:14.503Z    222.117.144.19
jekgru  2014-02-19T14:23:14.503Z    213.56.157.109
zmembx  2014-02-19T14:23:12.503Z    199.188.198.91
qhixcg  2014-02-19T14:23:11.503Z    203.40.104.119

and my code till now looks like this:

hijack_Reduced = FOREACH finalLogs GENERATE ClientUserName, timestamp, OriginalClientIP;
hijack_Filtered = FILTER hijack_Reduced BY OriginalClientIP != '-';

hijack_Sessionized = FOREACH (GROUP hijack_Filtered BY ClientUserName) {
  views = ORDER hijack_Filtered BY timestamp;
  GENERATE FLATTEN(Sessionize(views)) AS (ClientUserName,timestamp,OriginalClientIP,session_id);
}

but when i run this script, i got the following error Message:

15:36:22 ERROR - org.apache.pig.tools.pigstats.SimplePigStats.setBackendException(542) | ERROR 0: Exception while executing [POUserFunc (Name: POUserFunc(datafu.pig.sessions.Sessionize)[bag] - scope-199 Operator Key: scope-199) children: null at []]: java.lang.IllegalArgumentException: Invalid format: "aoebcu"

i already tried a lot, but nothing worked. do you got an idea?

Regards

Upvotes: 2

Views: 1155

Answers (1)

matterhayes
matterhayes

Reputation: 458

While you could write a UDF for this, you can actually make use of the UDFs already available in Apache DataFu to solve this.

My solution involves applying sessionization to the data. Basically you look at consecutive events and assign each event a session ID. If the time elapsed between two events exceeds a specified amount of time, in your case 5 seconds, then the next event gets a new session ID. Otherwise consecutive events get the same session ID. Once each event is assigned its session ID the rest is easy. We group by session ID and look for sessions that have more than one distinct IP address.

I'll walk through my solution.

Suppose you have the following input data. Both Harold and Kumar change their IP addresses. But Harold does it within 5 seconds, while Kumar does not. So the output of our script should just be simply "Harold".

  Harold,2014-02-18T16:14:49.503Z,123.123.123.123
  Harold,2014-02-18T16:14:51.503Z,123.123.123.123
  Harold,2014-02-18T16:14:55.503Z,321.321.321.321
  Kumar,2014-02-18T16:14:49.503Z,123.123.123.123
  Kumar,2014-02-18T16:14:55.503Z,123.123.123.123
  Kumar,2014-02-18T16:15:05.503Z,321.321.321.321

Load the data

data = LOAD 'input' using PigStorage(',') 
       AS (user:chararray,time:chararray,ip:chararray);

Now define a couple UDFs from DataFu. The Sessionize UDF performs sessionization as I described earlier. The DistinctBy UDF will be used to find the distinct IP addresses within each session.

define Sessionize datafu.pig.sessions.Sessionize('5s');

define DistinctBy datafu.pig.bags.DistinctBy('1');

Group the data by user, sort by time, and apply the Sessonize UDF. Note that the timestamp must be the first field, as this is what Sessionize expects. This UDF appends a session ID to each tuple. 

data = FOREACH data GENERATE time,user,ip;

data_sessionized = FOREACH (GROUP data BY user) {
  views = ORDER data BY time;
  GENERATE flatten(Sessionize(views)) as (time,user,ip,session_id);
}

Now that the data is sessionized, we can group by the user and session. I group by user too because I want to spit this value back out. We pass the bag of events into the DistinctBy UDF. Check the documentation of this UDF for a more detailed description. But essentially we will get as many tuples as there are distinct IP addresses per session. Note that I have removed the time from the relation below. This is because 1) it isn't needed, and 2) the DistinctBy in 1.2.0 of DataFu has a bug when handling fields containing dashes, as the time field does.

data_sessionized = FOREACH data_sessionized GENERATE user,ip,session_id;

data_sessionized = FOREACH (GROUP data_sessionized BY (user, session_id)) GENERATE
  group.user as user,
  SIZE(DistinctBy(data_sessionized)) as distinctIpCount;

Now select all the sessions that had more than one distinct IP address and return the distinct users for these sessions.

data_sessionized = FILTER data_sessionized BY distinctIpCount > 1;

data_sessionized = FOREACH data_sessionized GENERATE user;

data_sessionized = DISTINCT data_sessionized;

This produces simply:

Harold

Here is the full source code, which you should be able to paste directly into the DataFu unit tests and run:

  /**
  define Sessionize datafu.pig.sessions.Sessionize('5s');

  define DistinctBy datafu.pig.bags.DistinctBy('1'); -- distinct by ip

  data = LOAD 'input' using PigStorage(',') AS (user:chararray,time:chararray,ip:chararray);

  data = FOREACH data GENERATE time,user,ip;

  data_sessionized = FOREACH (GROUP data BY user) {
    views = ORDER data BY time;
    GENERATE flatten(Sessionize(views)) as (time,user,ip,session_id);
  }

  data_sessionized = FOREACH data_sessionized GENERATE user,ip,session_id;

  data_sessionized = FOREACH (GROUP data_sessionized BY (user, session_id)) GENERATE
    group.user as user,
    SIZE(DistinctBy(data_sessionized)) as distinctIpCount;

  data_sessionized = FILTER data_sessionized BY distinctIpCount > 1;

  data_sessionized = FOREACH data_sessionized GENERATE user;

  data_sessionized = DISTINCT data_sessionized;

  STORE data_sessionized INTO 'output';
   */
  @Multiline private String sessionizeUserIpTest;

  private String[] sessionizeUserIpTestData = new String[] {
      "Harold,2014-02-18T16:14:49.503Z,123.123.123.123",
      "Harold,2014-02-18T16:14:51.503Z,123.123.123.123",
      "Harold,2014-02-18T16:14:55.503Z,321.321.321.321",
      "Kumar,2014-02-18T16:14:49.503Z,123.123.123.123",
      "Kumar,2014-02-18T16:14:55.503Z,123.123.123.123",
      "Kumar,2014-02-18T16:15:05.503Z,321.321.321.321"
  };

  @Test
  public void sessionizeUserIpTest() throws Exception
  {
    PigTest test = createPigTestFromString(sessionizeUserIpTest);

    this.writeLinesToFile("input", 
        sessionizeUserIpTestData);

    List<Tuple> result = this.getLinesForAlias(test, "data_sessionized");

    assertEquals(result.size(),1);
    assertEquals(result.get(0).get(0),"Harold");
  }

Upvotes: 1

Related Questions