Skip to content

Commit 329eb6c

Browse files
venkateshkottapallivenkateshkottapalli
committed
Regex Parser example
1 parent 018db27 commit 329eb6c

7 files changed

Lines changed: 388 additions & 0 deletions

File tree

‎tutorials/parser/src/main/java/com/datatorrent/tutorial/regexparser/RegexParserApplication.java‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
package com.datatorrent.tutorial.regexparser;
2+
3+
import org.apache.hadoop.conf.Configuration;
4+
5+
import com.datatorrent.api.Context;
6+
import com.datatorrent.api.DAG;
7+
import com.datatorrent.api.StreamingApplication;
8+
import com.datatorrent.api.annotation.ApplicationAnnotation;
9+
import com.datatorrent.contrib.parser.RegexParser;
10+
import com.datatorrent.lib.appdata.schemas.SchemaUtils;
11+
import com.datatorrent.lib.io.ConsoleOutputOperator;
12+
import com.datatorrent.tutorial.csvparser.FileOutputOperator;
13+
14+
@ApplicationAnnotation(name = "RegexParser")
15+
public class RegexParserApplication implements StreamingApplication
16+
{
17+
@Override
18+
public void populateDAG(DAG dag, Configuration conf)
19+
{
20+
ServerLogGenerator logGenerator = dag.addOperator("logGenerator", ServerLogGenerator.class);
21+
RegexParser regexParser = dag.addOperator("regexParser", RegexParser.class);
22+
dag.setOutputPortAttribute(regexParser.out, Context.PortContext.TUPLE_CLASS, ServerLog.class);
23+
FileOutputOperator regexWriter = dag.addOperator("regexWriter", FileOutputOperator.class);
24+
FileOutputOperator regexErrorWriter = dag.addOperator("regexErrorWriter", FileOutputOperator.class);
25+
26+
dag.addStream("regexInput", logGenerator.outputPort, regexParser.in);
27+
dag.addStream("regexOutput", regexParser.out, regexWriter.input);
28+
dag.addStream("regexError", regexParser.err, regexErrorWriter.input);
29+
}
30+
}

‎tutorials/parser/src/main/java/com/datatorrent/tutorial/regexparser/ServerLog.java‎

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,99 @@
1+
package com.datatorrent.tutorial.regexparser;
2+
3+
import java.util.Date;
4+
5+
public class ServerLog
6+
{
7+
private Date date;
8+
private int id;
9+
private String signInId;
10+
private String ipAddress;
11+
private String serviceId;
12+
private String accountId;
13+
private String platform;
14+
15+
16+
public int getId()
17+
{
18+
return id;
19+
}
20+
21+
public void setId(int id)
22+
{
23+
this.id = id;
24+
}
25+
26+
public Date getDate()
27+
{
28+
return date;
29+
}
30+
31+
public void setDate(Date date)
32+
{
33+
this.date = date;
34+
}
35+
36+
public String getSignInId()
37+
{
38+
return signInId;
39+
}
40+
41+
public void setSignInId(String signInId)
42+
{
43+
this.signInId = signInId;
44+
}
45+
46+
public String getIpAddress()
47+
{
48+
return ipAddress;
49+
}
50+
51+
public void setIpAddress(String ipAddress)
52+
{
53+
this.ipAddress = ipAddress;
54+
}
55+
56+
public String getServiceId()
57+
{
58+
return serviceId;
59+
}
60+
61+
public void setServiceId(String serviceId)
62+
{
63+
this.serviceId = serviceId;
64+
}
65+
66+
public String getAccountId()
67+
{
68+
return accountId;
69+
}
70+
71+
public void setAccountId(String accountId)
72+
{
73+
this.accountId = accountId;
74+
}
75+
76+
public String getPlatform()
77+
{
78+
return platform;
79+
}
80+
81+
public void setPlatform(String platform)
82+
{
83+
this.platform = platform;
84+
}
85+
86+
@Override
87+
public String toString()
88+
{
89+
return "ServerLog{" +
90+
"date=" + date +
91+
", id=" + id +
92+
", signInId='" + signInId + '\'' +
93+
", ipAddress='" + ipAddress + '\'' +
94+
", serviceId='" + serviceId + '\'' +
95+
", accountId='" + accountId + '\'' +
96+
", platform='" + platform + '\'' +
97+
'}';
98+
}
99+
}

‎tutorials/parser/src/main/java/com/datatorrent/tutorial/regexparser/ServerLogGenerator.java‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
package com.datatorrent.tutorial.regexparser;
2+
3+
import com.datatorrent.api.DefaultOutputPort;
4+
import com.datatorrent.api.InputOperator;
5+
import com.datatorrent.common.util.BaseOperator;
6+
7+
public class ServerLogGenerator extends BaseOperator implements InputOperator
8+
{
9+
public transient DefaultOutputPort<byte[]> outputPort = new DefaultOutputPort<byte[]>();
10+
private int tupleRate = 10;
11+
private transient int tuplesEmmitedinWindow = 0;
12+
public int getTupleRate()
13+
{
14+
return tupleRate;
15+
}
16+
17+
public void setTupleRate(int tupleRate)
18+
{
19+
this.tupleRate = tupleRate;
20+
}
21+
22+
@Override
23+
public void emitTuples()
24+
{
25+
26+
while (tuplesEmmitedinWindow < tupleRate) {
27+
String line = "2015-10-01T03:14:49.000-07:00 lvn-d1-dev DevServer[9876]: INFO: [EVENT][SEQ=248717]" +
28+
" 2015:10:01:03:14:49 101 sign-in_id=11111@psop.com ip_address=1.1.1.1 service_id=IP1234-NPB12345_00 " +
29+
"result=RESULT_SUCCESconsole_id=0000000138e91b4e58236bf32besdafasdfasdfasdfsadf account_id=11111 platform=pik";
30+
outputPort.emit(line.getBytes());
31+
tuplesEmmitedinWindow++;
32+
}
33+
34+
}
35+
36+
@Override
37+
public void endWindow()
38+
{
39+
tuplesEmmitedinWindow = 0;
40+
super.endWindow();
41+
}
42+
}

‎tutorials/parser/src/main/resources/META-INF/properties.xml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,4 +37,8 @@
3737
</name>
3838
<value>Error:%s</value>
3939
</property>
40+
<property>
41+
<name>dt.application.csvParseApplication.operator.csvParser.prop.schema</name>
42+
<value>{"separator": ",","quoteChar":"\"","fields": [{"name": "adId","type": "Integer","constraints": {"required": "true"}},{"name": "campaignId","type": "Integer"},{"name": "adName","type": "String","constraints": {"required": "true","pattern": "^[a-zA-Z]*$","maxLength": "10"}},{"name": "bidPrice","type": "Double","constraints": {"required": "true","minValue": "0.1","maxValue": "3.2"}},{"name": "startDate","type": "Date","constraints": {"format": "yyyy-MM-dd HH:mm:ss","locale":"en"}},{"name": "endDate","type": "Date","constraints": {"format": "dd/MM/yyyy"}},{"name": "securityCode","type": "Long","constraints": {"minValue": "10","maxValue": "30"}},{"name": "active","type": "Boolean","constraints": {"required": "true"}},{"name": "optimized","type": "Boolean","constraints": { "trueValue":"OPTIMIZE","falseValue":"NO_OPTIMIZE"}},{"name": "parentCampaign","type": "String","constraints": {"required": "true","equals": "CAMP_add"}}]}</value>
43+
</property>
4044
</configuration>

‎tutorials/parser/src/site/conf/properties-regexParserApplication.xml‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
<?xml version="1.0"?>
2+
<configuration>
3+
4+
<property>
5+
<name>dt.application.RegexParser.operator.regexParser.prop.schema</name>
6+
<value>{
7+
"fields": [
8+
{
9+
"name": "date",
10+
"type": "Date",
11+
"constraints": {
12+
"format": "yyyy:MM:dd:hh:mm:ss"
13+
}
14+
},
15+
{
16+
"name": "id",
17+
"type": "Integer"
18+
},
19+
{
20+
"name": "signInId",
21+
"type": "String"
22+
23+
},
24+
{
25+
"name": "ipAddress",
26+
"type": "String"
27+
},
28+
{
29+
"name": "serviceId",
30+
"type": "String"
31+
},
32+
{
33+
"name": "accountId",
34+
"type": "String"
35+
},
36+
{
37+
"name": "platform",
38+
"type": "String"
39+
}
40+
]
41+
}
42+
</value>
43+
</property>
44+
<property>
45+
<name>dt.application.RegexParser.operator.regexParser.port.out.attr.TUPLE_CLASS</name>
46+
<value>com.datatorrent.tutorial.regexparser.ServerLog</value>
47+
</property>
48+
49+
<property>
50+
<name>dt.application.RegexParser.operator.regexParser.prop.splitRegexPattern</name>
51+
<value>.+\[SEQ=\w+\]\s*(\d+:[\d\d:]+)\s(\d+)\s* sign-in_id=(\S+) .*ip_address=(\S+).* service_id=(\S+).*account_id=(\S+).*platform=(\S+)</value>
52+
</property>
53+
54+
<property>
55+
<name>dt.application.RegexParser.operator.*.prop.filePath</name>
56+
<value>/tmp/application/parser/regexparser</value>
57+
</property>
58+
59+
<property>
60+
<name>dt.application.RegexParser.operator.regexErrorWriter.prop.outputFileName</name>
61+
<value>errordata</value>
62+
</property>
63+
64+
<property>
65+
<name>dt.application.RegexParser.operator.regexWriter.prop.outputFileName</name>
66+
<value>outputdata</value>
67+
</property>
68+
</configuration>

‎tutorials/parser/src/test/java/com/datatorrent/tutorial/regexparser/RegexParserApplicationTest.java‎

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
1+
package com.datatorrent.tutorial.regexparser;
2+
3+
import java.io.File;
4+
import java.io.IOException;
5+
import java.util.Collection;
6+
import java.util.concurrent.Callable;
7+
8+
import javax.validation.ConstraintViolationException;
9+
10+
import org.junit.Assert;
11+
import org.junit.Test;
12+
13+
import org.apache.commons.io.FileUtils;
14+
import org.apache.commons.io.filefilter.WildcardFileFilter;
15+
import org.apache.hadoop.conf.Configuration;
16+
17+
import com.datatorrent.api.LocalMode;
18+
import com.datatorrent.stram.StramLocalCluster;
19+
20+
public class RegexParserApplicationTest
21+
{
22+
23+
@Test
24+
public void testApplication() throws IOException, Exception
25+
{
26+
try {
27+
LocalMode lma = LocalMode.newInstance();
28+
Configuration conf = new Configuration(false);
29+
conf.addResource(this.getClass().getResourceAsStream("/properties-regexParserApplication.xml"));
30+
conf.setLong("dt.application.RegexParser.operator.logGenerator.prop.tupleRate", 10);
31+
final String dataFolderPath = conf.get("dt.application.RegexParser.operator.*.prop.filePath");
32+
final String dataFileName = conf
33+
.get("dt.application.RegexParser.operator.regexWriter.prop.outputFileName");
34+
35+
FileUtils.deleteDirectory(new File(dataFolderPath));
36+
lma.prepareDAG(new RegexParserApplication(), conf);
37+
LocalMode.Controller lc = lma.getController();
38+
lc.setHeartbeatMonitoringEnabled(false);
39+
((StramLocalCluster)lc).setExitCondition(new Callable<Boolean>()
40+
{
41+
@Override
42+
public Boolean call() throws Exception
43+
{
44+
if (new File(dataFolderPath).exists()) {
45+
Collection<File> files = FileUtils.listFiles(new File(dataFolderPath), new WildcardFileFilter(dataFileName
46+
+ "*"), null);
47+
if (files.size() >= 1) {
48+
File parsedFile = files.iterator().next();
49+
String fileData = FileUtils.readFileToString(parsedFile);
50+
String[] regexData = fileData.split("\n");
51+
return regexData.length == 10;
52+
}
53+
}
54+
return false;
55+
}
56+
});
57+
58+
lc.run(30 * 1000); // runs for 30 seconds and quits
59+
60+
Collection<File> files = FileUtils.listFiles(new File(dataFolderPath),
61+
new WildcardFileFilter(dataFileName + "*"), null);
62+
File parsedFile = files.iterator().next();
63+
String fileData = FileUtils.readFileToString(parsedFile);
64+
String[] logData = fileData.split("\n");
65+
for (String logLine : logData) {
66+
System.out.println(logLine);
67+
Assert.assertTrue(logLine.contains("id=" + 101));
68+
Assert.assertTrue(logLine.contains("signInId=" + "'11111@psop.com'"));
69+
Assert.assertTrue(logLine.contains("serviceId=" + "'IP1234-NPB12345_00'"));
70+
Assert.assertTrue(logLine.contains("accountId=" + "'11111'"));
71+
Assert.assertTrue(logLine.contains("platform=" + "'pik'"));
72+
}
73+
} catch (ConstraintViolationException e) {
74+
Assert.fail("constraint violations: " + e.getConstraintViolations());
75+
}
76+
}
77+
}

‎tutorials/parser/src/test/resources/properties-regexParserApplication.xml‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
<?xml version="1.0"?>
2+
<configuration>
3+
4+
<property>
5+
<name>dt.application.RegexParser.operator.regexParser.prop.schema</name>
6+
<value>{
7+
"fields": [
8+
{
9+
"name": "date",
10+
"type": "Date",
11+
"constraints": {
12+
"format": "yyyy:MM:dd:hh:mm:ss"
13+
}
14+
},
15+
{
16+
"name": "id",
17+
"type": "Integer"
18+
},
19+
{
20+
"name": "signInId",
21+
"type": "String"
22+
23+
},
24+
{
25+
"name": "ipAddress",
26+
"type": "String"
27+
},
28+
{
29+
"name": "serviceId",
30+
"type": "String"
31+
},
32+
{
33+
"name": "accountId",
34+
"type": "String"
35+
},
36+
{
37+
"name": "platform",
38+
"type": "String"
39+
}
40+
]
41+
}
42+
</value>
43+
</property>
44+
<property>
45+
<name>dt.application.RegexParser.operator.regexParser.port.out.attr.TUPLE_CLASS</name>
46+
<value>com.datatorrent.tutorial.regexparser.ServerLog</value>
47+
</property>
48+
49+
<property>
50+
<name>dt.application.RegexParser.operator.regexParser.prop.splitRegexPattern</name>
51+
<value>.+\[SEQ=\w+\]\s*(\d+:[\d\d:]+)\s(\d+)\s* sign-in_id=(\S+) .*ip_address=(\S+).* service_id=(\S+).*account_id=(\S+).*platform=(\S+)</value>
52+
</property>
53+
54+
<property>
55+
<name>dt.application.RegexParser.operator.*.prop.filePath</name>
56+
<value>/tmp/application/parser/regexparser</value>
57+
</property>
58+
59+
<property>
60+
<name>dt.application.RegexParser.operator.regexErrorWriter.prop.outputFileName</name>
61+
<value>errordata</value>
62+
</property>
63+
64+
<property>
65+
<name>dt.application.RegexParser.operator.regexWriter.prop.outputFileName</name>
66+
<value>outputdata</value>
67+
</property>
68+
</configuration>

0 commit comments

Comments
 (0)