Added log4j logging

Author: Maxim Moinat

Readme.md

@@ -1,4 +1,9 @@
-To be run with:
+Run manually with:
 ```
 hadoop jar restructurehdfs-all-0.1.0.jar webhdfs://radar-test.thehyve.net:50070 /topicAndroidPhoneNew/ output2
+```
+
+Cron example, executing every hour (/etc/crontab):
+```
+00 *    * * *   root    hadoop jar /home/maxim/hadoop-test/restructurehdfs-all-0.1.0.jar webhdfs://radar-test.thehyve.net:50070/ /udooE4Time/ /home/maxim/hadoop-test/outputCron > /home/maxim/hadoop-test/logCron.txt
 ```

build.gradle

@@ -12,6 +12,7 @@ repositories {
 
 dependencies {
     testCompile group: 'junit', name: 'junit', version: '4.11'
+    compile 'org.apache.logging.log4j:log4j-core:2.8.1'
     compile 'org.apache.hadoop:hadoop-common:2.7.3'
     runtime 'org.apache.hadoop:hadoop-hdfs:2.7.3'
     compile 'org.apache.avro:avro:1.7.7'
@@ -24,7 +25,7 @@ task fatJar(type: Jar) {
     manifest {
         attributes 'Implementation-Title': 'Gradle Jar File Example',
                 'Implementation-Version': version,
-                'Main-Class': 'org.radarcns.restructureAvroRecords'
+                'Main-Class': 'org.radarcns.RestructureAvroRecords'
     }
     baseName = project.name + '-all'
     from { configurations.compile.collect { it.isDirectory() ? it : zipTree(it) } }

src/main/java/org/radarcns/Frequency.java

@@ -3,11 +3,14 @@
 import org.apache.commons.collections.MapIterator;
 import org.apache.commons.collections.keyvalue.MultiKey;
 import org.apache.commons.collections.map.MultiKeyMap;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import java.io.*;
 
 
 public class Frequency {
+    private static final Logger logger = LogManager.getLogger(RestructureAvroRecords.class);
 
     private MultiKeyMap bins = new MultiKeyMap();
     private String binFilePath;
@@ -31,7 +34,7 @@ public void addToBin(String topicName, String id, String timestamp, Integer coun
 
     public void addToBin(String topicName, String id, Double time, Integer countToAdd) {
         // Hour resolution
-        String hourlyTimestamp = restructureAvroRecords.createHourTimestamp(time);
+        String hourlyTimestamp = RestructureAvroRecords.createHourTimestamp(time);
 
         addToBin(topicName, id, hourlyTimestamp, countToAdd);
     }
@@ -91,9 +94,9 @@ private void addBinsFromFile() {
                 this.addToBin(columns[0], columns[1], columns[2], Integer.valueOf(columns[3]));
             }
         } catch (IOException e) {
-            System.out.println("Could not read the file with bins. Creating new file when writing.");
+            logger.warn("Could not read the file with bins. Creating new file when writing.");
         } catch (ArrayIndexOutOfBoundsException e) {
-            System.out.println("Unable to parse the contents of the bins file. Skipping reading.");
+            logger.warn("Unable to parse the contents of the bins file. Skipping reading.");
         }
     }
 }

src/main/java/org/radarcns/restructureAvroRecords.java

@@ -11,12 +11,15 @@
 import org.apache.hadoop.fs.LocatedFileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import java.io.*;
 import java.text.SimpleDateFormat;
 import java.util.*;
 
-public class restructureAvroRecords {
+public class RestructureAvroRecords {
+    private static final Logger logger = LogManager.getLogger(RestructureAvroRecords.class);
 
     private final String OUTPUT_FILE_EXTENSION = "json";
     private final String OFFSETS_FILE_NAME = "offsets.csv";
@@ -34,23 +37,28 @@ public class restructureAvroRecords {
     private int processedRecordsCount;
 
     public static void main(String [] args) throws Exception {
-        restructureAvroRecords restr = new restructureAvroRecords(args[0], args[2]);
+        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+
+        logger.info(dateFormat.format(new Date()));
+        logger.info("Starting...");
+        logger.info("In:  " + args[0] + args[1]);
+        logger.info("Out: " + args[2]);
+
         long time1 = System.currentTimeMillis();
+
+        RestructureAvroRecords restr = new RestructureAvroRecords(args[0], args[2]);
         restr.start(args[1]);
-        System.out.printf("Processed %d files and %,d records\n", restr.getProcessedFileCount(), restr.getProcessedRecordsCount());
-        System.out.printf("Time taken: %.2f seconds\n",(System.currentTimeMillis() - time1)/1000d);
 
-//        restructureAvroRecords restr = new restructureAvroRecords("webhdfs://radar-test.thehyve.net:50070", "output4/");
-//        restr.start("/topicE4/");
+        logger.info(String.format("Processed %d files and %,d records", restr.getProcessedFileCount(), restr.getProcessedRecordsCount()));
+        logger.info(String.format("Time taken: %.2f seconds",(System.currentTimeMillis() - time1)/1000d));
 
 //        restr.processTopic(new Path("/topicE4/android_empatica_e4_temperature/"));
-//        restr.processAvroFile(new Path("/testE4Time/android_phone_acceleration/partition=0/android_phone_acceleration+0+0000590000+0000599999.avro"),"wazaa" );
+//        restr.processFile(new Path("/testE4Time/android_phone_acceleration/partition=0/android_phone_acceleration+0+0000590000+0000599999.avro"),"wazaa" );
     }
 
-    public restructureAvroRecords(String inputPath, String outputPath) {
+    public RestructureAvroRecords(String inputPath, String outputPath) {
         this.setInputWebHdfsURL(inputPath);
         this.setOutputPath(outputPath);
-        bins.setBinFilePath(outputPath + "/" + BINS_FILE_NAME);
     }
 
     public void setInputWebHdfsURL(String fileSystemURL) {
@@ -61,6 +69,7 @@ public void setOutputPath(String path) {
         // Remove trailing backslash
         outputPath = path.replaceAll("/$","");
         offsetsPath = outputPath + "/" + OFFSETS_FILE_NAME;
+        bins.setBinFilePath(outputPath + "/" + BINS_FILE_NAME);
     }
 
     public int getProcessedFileCount() {
@@ -107,16 +116,16 @@ private void processTopic(Path topicPath) throws IOException {
             LocatedFileStatus locatedFileStatus = files.next();
 
             if (locatedFileStatus.isFile())
-                this.processAvroFile( locatedFileStatus.getPath(), topicName );
+                this.processFile(locatedFileStatus.getPath(), topicName);
         }
     }
 
-    private void processAvroFile(Path filePath, String topicName) throws IOException {
+    private void processFile(Path filePath, String topicName) throws IOException {
         String fileName = filePath.getName();
 
         // Skip if extension is not .avro
         if (! FilenameUtils.getExtension(fileName).equals("avro")) {
-            System.out.printf("Skipped non avro file: %s\n", fileName);
+            logger.info("Skipped non-avro file: " + fileName);
             return;
         }
 
@@ -125,11 +134,12 @@ private void processAvroFile(Path filePath, String topicName) throws IOException
             return;
         }
 
-        System.out.println(filePath);
+        logger.info(filePath);
+
         // Read and parse avro file
         FsInput input = new FsInput(filePath, conf);
-        DatumReader<GenericRecord> datumReader = new GenericDatumReader<GenericRecord>();
-        DataFileReader<GenericRecord> dataFileReader = new DataFileReader<GenericRecord>(input, datumReader);
+        DatumReader<GenericRecord> datumReader = new GenericDatumReader<>();
+        DataFileReader<GenericRecord> dataFileReader = new DataFileReader<>(input, datumReader);
 
         GenericRecord record = null;
         while (dataFileReader.hasNext()) {
@@ -176,9 +186,9 @@ private void appendToFile(String directoryName, String fileName, String data) {
         File directory = new File(directoryName);
         if (! directory.exists()){
             if (directory.mkdirs())
-                System.out.printf("Created directory: %s\n", directory.getAbsolutePath());
+                logger.info("Created directory: " + directory.getAbsolutePath());
             else
-                System.out.printf("FAILED to create directory: %s\n", directory.getAbsolutePath());
+                logger.warn("FAILED to create directory: " + directory.getAbsolutePath());
         }
 
         String filePath = directoryName + "/" + fileName;
@@ -206,10 +216,10 @@ private void writeSeenOffsets(String fileName) {
             fromOffset = Integer.valueOf( fileNameParts[2] );
             toOffset = Integer.valueOf( fileNameParts[3] );
         } catch (IndexOutOfBoundsException e) {
-            System.out.println("Could not split filename to the commit offsets.");
+            logger.warn("Could not split filename to the commit offsets.");
             return;
         } catch (NumberFormatException e) {
-            System.out.println("Could not convert offsets to integers.");
+            logger.warn("Could not convert offsets to integers.");
             return;
         }
 
@@ -236,9 +246,8 @@ private void readSeenOffsets() {
                 String[] columns = line.split(",");
                 seenFiles.add(columns[0]);
             }
-
         } catch (IOException e) {
-            System.out.println("No files processed yet.");
+            logger.info("Offsets file does not exist yet, will be created.");
         }
     }
 }

src/main/resources/log4j2.xml

@@ -0,0 +1,17 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="INFO">
+    <Appenders>
+        <Console name="Console" target="SYSTEM_OUT">
+            <PatternLayout pattern="%d{HH:mm:ss.SSS} %-5level %logger{36} - %msg%n" />
+        </Console>
+        <File name="MyFile" fileName="restructureAvroRecords.log" immediateFlush="true" append="true">
+            <PatternLayout pattern="%d{yyy-MM-dd HH:mm:ss.SSS} %-5level %logger{36} - %msg%n"/>
+        </File>
+    </Appenders>
+    <Loggers>
+        <Root level="debug">
+            <AppenderRef ref="Console" />
+            <AppenderRef ref="MyFile"/>
+        </Root>
+    </Loggers>
+</Configuration>