Index: modules/dnet-openaire-usage-stats-export/trunk/src/main/java/eu/dnetlib/usagestats/export/SushiStats.java
===================================================================
--- modules/dnet-openaire-usage-stats-export/trunk/src/main/java/eu/dnetlib/usagestats/export/SushiStats.java	(nonexistent)
+++ modules/dnet-openaire-usage-stats-export/trunk/src/main/java/eu/dnetlib/usagestats/export/SushiStats.java	(revision 47499)
@@ -0,0 +1,280 @@
+package eu.dnetlib.usagestats.export;
+
+import java.io.*;
+//import java.io.BufferedReader;
+//import java.io.InputStreamReader;
+import java.net.URL;
+import java.net.URLConnection;
+import java.sql.ResultSet;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Calendar;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.Statement;
+
+import org.json.simple.JSONArray;
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
+
+import org.apache.log4j.Logger;
+
+/**
+ * Created by tsampikos on 28/3/2017.
+ */
+public class SushiStats {
+     private final String dbUrl;
+    private final String dbSchema;
+    private final String dbUserName;
+    private final String dbPassword;
+
+    private Connection conn = null;
+    private Statement stmt = null;
+
+    private final Logger log = Logger.getLogger(this.getClass());
+
+    public SushiStats(String dbUrl, String dbUsername, String dbPassword) throws Exception {
+        this.dbUrl = dbUrl;
+        this.dbSchema = "shadow";
+        this.dbUserName = dbUsername;
+        this.dbPassword = dbPassword;
+
+        connectDB();
+        createTables();
+    }
+
+    private void connectDB() throws Exception {
+        try {
+            Class.forName("org.postgresql.Driver");
+            conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword);
+
+            stmt = conn.createStatement();
+            String sqlSetSearchPath = "SET search_path TO " + dbSchema + ";";
+            stmt.executeUpdate(sqlSetSearchPath);
+
+            log.info("Opened database successfully");
+
+        } catch (Exception e) {
+            log.error("Connect to db failed: " + e);
+            throw new Exception("Failed to connect to db: " + e.toString(), e);
+        }
+    }
+
+    private void createTables() throws Exception {
+        try {
+
+            stmt = conn.createStatement();
+            String sqlCreateTableSushiLog = "CREATE TABLE IF NOT EXISTS sushilog(source TEXT, repository TEXT, oai TEXT, date TEXT, metric_type TEXT, count INT, PRIMARY KEY(source, repository, oai, date, metric_type));";
+            stmt.executeUpdate(sqlCreateTableSushiLog);
+
+            String sqlCopyPublicSushiLog="INSERT INTO sushilog SELECT * FROM public.sushilog;";
+            stmt.executeUpdate(sqlCopyPublicSushiLog);
+
+            String sqlcreateRuleSushiLog = "CREATE OR REPLACE RULE ignore_duplicate_inserts AS " +
+                " ON INSERT TO sushilog " +
+                " WHERE (EXISTS ( SELECT sushilog.source, sushilog.repository," +
+                "sushilog.oai, sushilog.date " +
+                "FROM sushilog " +
+                "WHERE sushilog.source = new.source AND sushilog.repository = new.repository AND sushilog.oai = new.oai AND sushilog.date = new.date AND sushilog.metric_type = new.metric_type)) DO INSTEAD NOTHING;";
+            stmt.executeUpdate(sqlcreateRuleSushiLog);
+
+            stmt.close();
+            conn.close();
+            log.info("Sushi Tables Created");
+        } catch (Exception e) {
+            log.error("Failed to create tables: " + e);
+            throw new Exception("Failed to create tables: " + e.toString(), e);
+        }
+    }
+
+    public void sushiStats() throws Exception {
+        if (conn.isClosed())
+            connectDB();
+
+        stmt = conn.createStatement();
+        conn.setAutoCommit(false);
+
+        //String sql = "INSERT INTO sushi_result_downloads SELECT s.source, d.id AS repository, ro.id, s.date, s.count FROM sushilog s, datasource_oids d, result_oids ro WHERE s.repository=d.orid AND s.oai=ro.orid AND metric_type='ft_total'";
+        //String sql = "SELECT s.source, d.id AS repository_id, ro.id as result_id, extract('year' from s.date::date) ||'/'|| LPAD(CAST(extract('month' from s.date::date) AS VARCHAR), 2, '0') as date, s.count INTO downloads_stats FROM sushilog s, datasource_oids d, result_oids ro WHERE s.repository=d.orid AND s.oai=ro.orid AND metric_type='ft_total'";
+        //String sql = "INSERT INTO downloads_stats SELECT s.source, d.id AS repository_id, ro.id as result_id, extract('year' from s.date::date) ||'/'|| LPAD(CAST(extract('month' from s.date::date) AS VARCHAR), 2, '0') as date, s.count FROM sushilog s, datasource_oids d, result_oids ro WHERE s.repository=d.orid AND s.oai=ro.orid AND metric_type='ft_total';";
+        String sql = "INSERT INTO downloads_stats SELECT s.source, d.id AS repository_id, ro.id as result_id, extract('year' from s.date::date) ||'/'|| LPAD(CAST(extract('month' from s.date::date) AS VARCHAR), 2, '0') as date, s.count, '0' FROM sushilog s, datasource_oids d, result_oids ro WHERE s.repository=d.orid AND s.oai=ro.orid AND metric_type='ft_total';";
+        stmt.executeUpdate(sql);
+
+
+        stmt.close();
+        conn.commit();
+        conn.close();
+    }
+
+    public void processIrusRRReport() throws Exception {
+        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("YYYY-MM");
+        String reportUrl = "http://www.irus.mimas.ac.uk/api/sushilite/v1_7/GetReport/?Report=RR1&Release=4&RequestorID=OpenAIRE&BeginDate=2016-01&EndDate=" + simpleDateFormat.format(new Date()) + "&RepositoryIdentifier=&ItemDataType=&NewJiscBand=&Granularity=Monthly&Callback=";
+        String text = getJson(reportUrl, "", "");
+
+        JSONParser parser = new JSONParser();
+        JSONObject jsonObject = (JSONObject) parser.parse(text);
+        jsonObject = (JSONObject) jsonObject.get("ReportResponse");
+        jsonObject = (JSONObject) jsonObject.get("Report");
+        jsonObject = (JSONObject) jsonObject.get("Report");
+        jsonObject = (JSONObject) jsonObject.get("Customer");
+        JSONArray jsonArray = (JSONArray) jsonObject.get("ReportItems");
+        int i = 0;
+        for (Object aJsonArray : jsonArray) {
+            JSONObject jsonObjectRow = (JSONObject) aJsonArray;
+            JSONArray itemIdentifier = (JSONArray) jsonObjectRow.get("ItemIdentifier");
+            for (Object identifier : itemIdentifier) {
+                JSONObject opendoar = (JSONObject) identifier;
+                if(opendoar.get("Type").toString().equals("OpenDOAR")){
+                    //System.out.println(i + ": " + opendoar.get("Value").toString());
+                    log.info(i + ": " + opendoar.get("Value").toString());
+                    i++;
+                    processIrusIRReport(opendoar.get("Value").toString());
+                    break;
+                }
+            }
+            //break;
+        }
+    }
+
+    private void processIrusIRReport(String opendoar) throws Exception {
+        if (conn.isClosed())
+            connectDB();
+
+        conn.setAutoCommit(false);
+
+        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("YYYY-MM");
+
+        Calendar start = Calendar.getInstance();
+        start.set(Calendar.YEAR, 2016);
+        start.set(Calendar.MONTH, Calendar.JANUARY);
+        //start.setTime(simpleDateFormat.parse("2016-01"));
+
+        Calendar end = Calendar.getInstance();
+        end.add(Calendar.DAY_OF_MONTH, -1);
+
+        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
+        PreparedStatement st = conn.prepareStatement("SELECT max(date) FROM public.sushilog WHERE repository=?;");
+        st.setString(1, "opendoar____::" + opendoar);
+        ResultSet rs_date = st.executeQuery();
+        while(rs_date.next()){
+            if(rs_date.getString(1) != null && !rs_date.getString(1).equals("null") && !rs_date.getString(1).equals("")) {
+                start.setTime(sdf.parse(rs_date.getString(1)));
+            }
+        }
+        rs_date.close();
+
+        PreparedStatement preparedStatement = conn.prepareStatement("INSERT INTO sushilog (source, repository, oai, date, metric_type, count) VALUES (?,?,?,?,?,?)");
+        int batch_size = 0;
+
+        while(start.before(end)){
+            //log.info("date: " + simpleDateFormat.format(start.getTime()));
+            String reportUrl = "http://irus.mimas.ac.uk/api/sushilite/v1_7/GetReport/?Report=IR1&Release=4&RequestorID=OpenAIRE&BeginDate=" + simpleDateFormat.format(start.getTime()) + "&EndDate=" + simpleDateFormat.format(start.getTime()) + "&RepositoryIdentifier=opendoar%3A" + opendoar + "&ItemIdentifier=&ItemDataType=&hasDOI=&Granularity=Monthly&Callback=";
+            start.add(Calendar.MONTH, 1);
+
+            String text = getJson(reportUrl, "", "");
+            if(text == null){
+                continue;
+            }
+
+            JSONParser parser = new JSONParser();
+            JSONObject jsonObject = (JSONObject) parser.parse(text);
+            jsonObject = (JSONObject) jsonObject.get("ReportResponse");
+            jsonObject = (JSONObject) jsonObject.get("Report");
+            jsonObject = (JSONObject) jsonObject.get("Report");
+            jsonObject = (JSONObject) jsonObject.get("Customer");
+            JSONArray jsonArray = (JSONArray) jsonObject.get("ReportItems");
+            if(jsonArray == null){
+                continue;
+            }
+            String oai = "";
+            for (Object aJsonArray : jsonArray) {
+                JSONObject jsonObjectRow = (JSONObject) aJsonArray;
+                JSONArray itemIdentifier = (JSONArray) jsonObjectRow.get("ItemIdentifier");
+                for (Object identifier : itemIdentifier) {
+                    JSONObject oaiPmh = (JSONObject) identifier;
+                    if(oaiPmh.get("Type").toString().equals("OAI")){
+                        oai = oaiPmh .get("Value").toString();
+                        //System.out.println("OAI: " + oai);
+                        break;
+                    }
+                }
+
+                JSONArray itemPerformance = (JSONArray) jsonObjectRow.get("ItemPerformance");
+                String period;
+                String type;
+                String count;
+                for (Object perf : itemPerformance) {
+                    JSONObject performance = (JSONObject) perf;
+                    JSONObject periodObj = (JSONObject) performance.get("Period");
+                    period = periodObj.get("Begin").toString();
+                    JSONObject instanceObj = (JSONObject) performance.get("Instance");
+                    type = instanceObj.get("MetricType").toString();
+                    count = instanceObj.get("Count").toString();
+                    //System.out.println(oai + " : " + period + " : " + count);
+
+                    preparedStatement.setString(1, "IRUS-UK");
+                    preparedStatement.setString(2, "opendoar____::" + opendoar);
+                    preparedStatement.setString(3, oai);
+                    preparedStatement.setString(4, period);
+                    preparedStatement.setString(5, type);
+                    preparedStatement.setInt(6, Integer.parseInt(count));
+                    preparedStatement.addBatch();
+                    batch_size++;
+                    if(batch_size == 10000){
+                        preparedStatement.executeBatch();
+                        conn.commit();
+                        batch_size = 0;
+                    }
+                }
+                //break;
+            }
+            //break;
+        }
+
+        preparedStatement.executeBatch();
+        conn.commit();
+        conn.close();
+    }
+
+    private String getJson(String url,String username, String password) throws Exception {
+        //String cred=username+":"+password;
+        //String encoded = new sun.misc.BASE64Encoder().encode (cred.getBytes());
+        try {
+            URL website = new URL(url);
+            URLConnection connection = website.openConnection();
+            //connection.setRequestProperty ("Authorization", "Basic "+encoded);
+            StringBuilder response;
+            try (BufferedReader in = new BufferedReader(new InputStreamReader(connection.getInputStream()))) {
+                response = new StringBuilder();
+                String inputLine;
+                while ((inputLine = in.readLine()) != null) {
+                    response.append(inputLine);
+                    response.append("\n");
+                }
+            }
+            return response.toString();
+        }catch (Exception e){
+            log.error("Failed to get URL: " + e);
+            //System.out.println("Failed to get URL: " + e);
+            return null;
+            //throw new Exception("Failed to get URL: " + e.toString(), e);
+        }
+    }
+
+    /*
+    private void flushString(String data, String destination) throws Exception {
+        FSDataOutputStream fin;
+        try {
+            FileSystem fs = FileSystem.get(new Configuration());
+            fin = fs.create(new Path(destination), true);
+            fin.write(data.getBytes());
+            fin.close();
+        } catch (Exception e) {
+            log.error("Failed  to write exported data to a file : ", e);
+            throw new Exception("Failed  to write exported data to a file : " + e.toString(), e);
+        }
+    }
+    */
+}
Index: modules/dnet-openaire-usage-stats-export/trunk/src/main/java/eu/dnetlib/usagestats/export/PiwikStatsDB.java
===================================================================
--- modules/dnet-openaire-usage-stats-export/trunk/src/main/java/eu/dnetlib/usagestats/export/PiwikStatsDB.java	(nonexistent)
+++ modules/dnet-openaire-usage-stats-export/trunk/src/main/java/eu/dnetlib/usagestats/export/PiwikStatsDB.java	(revision 47499)
@@ -0,0 +1,1116 @@
+package eu.dnetlib.usagestats.export;
+
+import java.io.*;
+import java.net.URLDecoder;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.text.SimpleDateFormat;
+import java.util.*;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.log4j.Logger;
+import org.json.simple.JSONArray;
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
+
+public class PiwikStatsDB {
+    private final String dbUrl;
+    private final String dbSchema;
+    private final String dbUserName;
+    private final String dbPassword;
+    private final String logPath;
+
+    private Connection conn = null;
+    private Statement stmt = null;
+
+    private final Logger log = Logger.getLogger(this.getClass());
+
+    public PiwikStatsDB(String dbUrl, String dbUserName, String dbPassword, String logPath) throws Exception {
+        this.dbUrl = dbUrl;
+        this.dbSchema = "shadow";
+        this.dbUserName = dbUserName;
+        this.dbPassword = dbPassword;
+        this.logPath = logPath;
+
+        connectDB();
+        createTables();
+    }
+
+    private void connectDB() throws Exception {
+        try {
+            Class.forName("org.postgresql.Driver");
+            conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword);
+
+            stmt = conn.createStatement();
+            String sqlSetSearchPath = "SET search_path TO " + dbSchema + ";";
+            stmt.executeUpdate(sqlSetSearchPath);
+
+            //log.info("Opened database successfully");
+        } catch (Exception e) {
+            log.error("Connect to db failed: " + e);
+            throw new Exception("Failed to connect to db: " + e.toString(), e);
+        }
+    }
+
+    private void createTables() throws Exception {
+        try {
+            stmt = conn.createStatement();
+            String sqlCreateTablePiwikLog = "CREATE TABLE IF NOT EXISTS piwiklog(source INTEGER, id_visit TEXT, country TEXT, action TEXT, url TEXT, entity_id TEXT, source_item_type TEXT, timestamp TEXT, referrer_name TEXT, agent TEXT, PRIMARY KEY(source, id_visit, action, timestamp, entity_id));";
+            String sqlcreateRulePiwikLog = "CREATE OR REPLACE RULE ignore_duplicate_inserts AS " +
+                    " ON INSERT TO piwiklog " +
+                    " WHERE (EXISTS ( SELECT piwiklog.source, piwiklog.id_visit," +
+                    "piwiklog.action, piwiklog.\"timestamp\", piwiklog.entity_id " +
+                    "FROM piwiklog " +
+                    "WHERE piwiklog.source = new.source AND piwiklog.id_visit = new.id_visit AND piwiklog.action = new.action AND piwiklog.entity_id = new.entity_id AND piwiklog.\"timestamp\" = new.\"timestamp\")) DO INSTEAD NOTHING;";
+            stmt.executeUpdate(sqlCreateTablePiwikLog);
+            stmt.executeUpdate(sqlcreateRulePiwikLog);
+
+            String sqlCopyPublicPiwiklog="insert into piwiklog select * from public.piwiklog;";
+            stmt.executeUpdate(sqlCopyPublicPiwiklog);
+
+            String sqlCreateTablePortalLog = "CREATE TABLE IF NOT EXISTS process_portal_log(source INTEGER, id_visit TEXT, country TEXT, action TEXT, url TEXT, entity_id TEXT, source_item_type TEXT, timestamp TEXT, referrer_name TEXT, agent TEXT, PRIMARY KEY(source, id_visit, timestamp));";
+            String sqlcreateRulePortalLog = "CREATE OR REPLACE RULE ignore_duplicate_inserts AS " +
+                    " ON INSERT TO process_portal_log " +
+                    " WHERE (EXISTS ( SELECT process_portal_log.source, process_portal_log.id_visit," +
+                    "process_portal_log.\"timestamp\" " +
+                    "FROM process_portal_log " +
+                    "WHERE process_portal_log.source = new.source AND process_portal_log.id_visit = new.id_visit AND process_portal_log.\"timestamp\" = new.\"timestamp\")) DO INSTEAD NOTHING;";
+            stmt.executeUpdate(sqlCreateTablePortalLog);
+            stmt.executeUpdate(sqlcreateRulePortalLog);
+
+            stmt.close();
+            conn.close();
+            log.info("Usage Tables Created");
+
+        } catch (Exception e) {
+            log.error("Failed to create tables: " + e);
+            throw new Exception("Failed to create tables: " + e.toString(), e);
+            //System.exit(0);
+        }
+    }
+
+    protected void processLogs() throws Exception {
+        try {
+            processRepositoryLog();
+            log.info("repository process done");
+            removeDoubleClicks();
+            log.info("removing double clicks done");
+            cleanOAI();
+            log.info("cleaning oai done");
+
+            processPortalLog();
+            log.info("portal process done");
+            portalStats();
+            log.info("portal stats done");
+        } catch (Exception e) {
+            log.error("Failed to process logs: " + e);
+            throw new Exception("Failed to process logs: " + e.toString(), e);
+        }
+    }
+
+    protected void usageStats() throws Exception {
+        try {
+            //resultStats();
+            //dataSourceStats();
+            //organizationsStats();
+            //projectsStats();
+            //repositoryViewsStats();
+            //repositoryDownloadsStats();
+            viewsStats();
+            downloadsStats();
+            log.info("stat tables and views done");
+        } catch (Exception e) {
+            log.error("Failed to create usage stats: " + e);
+            throw new Exception("Failed to create usage stats: " + e.toString(), e);
+        }
+    }
+
+    // Import repository Logs to DB
+    private void processRepositoryLog() throws Exception {
+        if (conn.isClosed())
+            connectDB();
+
+        stmt = conn.createStatement();
+        conn.setAutoCommit(false);
+
+        ArrayList<String> jsonFiles = listHdfsDir(logPath + "repolog");
+
+        PreparedStatement prepStatem = conn.prepareStatement("INSERT INTO piwiklog (source, id_visit, country, action, url, entity_id, source_item_type, timestamp, referrer_name, agent) VALUES (?,?,?,?,?,?,?,?,?,?)");
+        int batch_size = 0;
+        for (String jsonFile : jsonFiles) {
+            JSONParser parser = new JSONParser();
+            JSONArray jsonArray = (JSONArray) parser.parse(readHDFSFile(jsonFile));
+
+            for (Object aJsonArray : jsonArray) {
+                JSONObject jsonObjectRow = (JSONObject) aJsonArray;
+                int idSite = Integer.parseInt(jsonObjectRow.get("idSite").toString());
+                String idVisit = jsonObjectRow.get("idVisit").toString();
+                String country = jsonObjectRow.get("country").toString();
+                String referrerName = jsonObjectRow.get("referrerName").toString();
+                String agent = jsonObjectRow.get("browser").toString();
+
+                String sourceItemType = "repItem";
+
+                JSONArray actionDetails = (JSONArray) jsonObjectRow.get(("actionDetails"));
+                for (Object actionDetail : actionDetails) {
+                    JSONObject actionDetailsObj = (JSONObject) actionDetail;
+
+                    if (actionDetailsObj.get("customVariables") != null) {
+                        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+                        simpleDateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));
+                        Timestamp timestamp = new Timestamp(Long.parseLong(actionDetailsObj.get("timestamp").toString()) * 1000);
+                        String url = actionDetailsObj.get("url").toString();
+                        String oaipmh = ((JSONObject) ((JSONObject) actionDetailsObj.get("customVariables")).get("1")).get("customVariablePageValue1").toString();
+                        String action = actionDetailsObj.get("type").toString();
+
+                        prepStatem.setInt(1, idSite);
+                        prepStatem.setString(2, idVisit);
+                        prepStatem.setString(3, country);
+                        prepStatem.setString(4, action);
+                        prepStatem.setString(5, url);
+                        prepStatem.setString(6, oaipmh);
+                        prepStatem.setString(7, sourceItemType);
+                        prepStatem.setString(8, simpleDateFormat.format(timestamp));
+                        prepStatem.setString(9, referrerName);
+                        prepStatem.setString(10, agent);
+                        prepStatem.addBatch();
+                        batch_size++;
+                        if (batch_size == 10000) {
+                            prepStatem.executeBatch();
+                            conn.commit();
+                            batch_size = 0;
+                        }
+                    }
+                }
+            }
+        }
+        prepStatem.executeBatch();
+        conn.commit();
+        stmt.close();
+        conn.close();
+    }
+
+    private void removeDoubleClicks() throws Exception {
+        if (conn.isClosed())
+            connectDB();
+
+        stmt = conn.createStatement();
+        conn.setAutoCommit(false);
+
+        //clean download double clicks
+        String sql = "DELETE FROM piwiklog p WHERE EXISTS (SELECT DISTINCT p1.source, p1.id_visit, p1.action, p1.entity_id, p1.timestamp FROM piwiklog p1, piwiklog p2 WHERE p1.source!='5' AND p1.source=p2.source AND p1.id_visit=p2.id_visit AND p1.entity_id=p2.entity_id AND p1.action=p2.action AND p1.action='download' AND p1.timestamp!=p2.timestamp AND p1.timestamp<p2.timestamp AND extract(EPOCH FROM p2.timestamp::timestamp-p1.timestamp::timestamp)<30 AND p.source=p1.source AND p.id_visit=p1.id_visit AND p.action=p1.action AND p.entity_id=p1.entity_id AND p.timestamp=p1.timestamp);";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+
+        //clean view double clicks
+        sql = "DELETE FROM piwiklog p WHERE EXISTS (SELECT DISTINCT p1.source, p1.id_visit, p1.action, p1.entity_id, p1.timestamp from piwiklog p1, piwiklog p2 WHERE p1.source!='5' AND p1.source=p2.source AND p1.id_visit=p2.id_visit AND p1.entity_id=p2.entity_id AND p1.action=p2.action AND p1.action='action' AND p1.timestamp!=p2.timestamp AND p1.timestamp<p2.timestamp AND extract(EPOCH FROM p2.timestamp::timestamp-p1.timestamp::timestamp)<10 AND p.source=p1.source AND p.id_visit=p1.id_visit AND p.action=p1.action AND p.entity_id=p1.entity_id AND p.timestamp=p1.timestamp);";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+        conn.close();
+    }
+
+    private void viewsStats() throws Exception {
+        if (conn.isClosed())
+            connectDB();
+
+        stmt = conn.createStatement();
+        conn.setAutoCommit(false);
+
+        //String sql = "CREATE OR REPLACE VIEW result_views_monthly AS SELECT entity_id AS id, COUNT(entity_id) as views, extract('year' from timestamp::date) ||'/'|| LPAD(CAST(extract('month' from timestamp::date) AS VARCHAR), 2, '0') AS month, source FROM piwiklog where action='action' and (source_item_type='oaItem' or source_item_type='repItem') group by id, month, source order by source, id, month;";
+        String sql = "CREATE OR REPLACE VIEW result_views_monthly AS SELECT entity_id AS id, COUNT(entity_id) as views, SUM(CASE WHEN referrer_name LIKE '%openaire%' THEN 1 ELSE 0 END) AS openaire_referrer, extract('year' from timestamp::date) ||'/'|| LPAD(CAST(extract('month' from timestamp::date) AS VARCHAR), 2, '0') AS month, source FROM piwiklog where action='action' and (source_item_type='oaItem' or source_item_type='repItem') group by id, month, source order by source, id, month;";
+        stmt.executeUpdate(sql);
+
+        sql = "SELECT 'OpenAIRE'::TEXT as source, d.id as repository_id, ro.id as result_id, month as date, max(views) AS count, max(openaire_referrer) AS openaire INTO views_stats FROM result_views_monthly p, datasource d, result_oids ro where p.source!='5' AND p.source=d.piwik_id and p.id=ro.orid group by repository_id, result_id, date ORDER BY repository_id, result_id, date;";
+        stmt.executeUpdate(sql);
+
+        sql = "SELECT 'OpenAIRE'::TEXT as source, d.id as repository_id, ro.id as result_id, month as date, max(views) AS count INTO pageviews_stats FROM result_views_monthly p, datasource d, result_oids ro where p.source='5' AND p.source=d.piwik_id and p.id=ro.orid group by repository_id, result_id, date ORDER BY repository_id, result_id, date;";
+        stmt.executeUpdate(sql);
+
+        sql = "DROP VIEW IF EXISTS result_views_monthly;";
+        stmt.executeUpdate(sql);
+
+        stmt.close();
+        conn.commit();
+        conn.close();
+    }
+
+    private void downloadsStats() throws Exception {
+        if (conn.isClosed())
+            connectDB();
+
+        stmt = conn.createStatement();
+        conn.setAutoCommit(false);
+
+        //String sql = "CREATE OR REPLACE VIEW result_downloads_monthly as select entity_id AS id, COUNT(entity_id) as downloads, extract('year' from timestamp::date) ||'/'|| LPAD(CAST(extract('month' from timestamp::date) AS VARCHAR), 2, '0') AS month, source FROM piwiklog where action='download' and (source_item_type='oaItem' or source_item_type='repItem') group by id, month, source order by source, id, month;";
+        String sql = "CREATE OR REPLACE VIEW result_downloads_monthly as select entity_id AS id, COUNT(entity_id) as downloads, SUM(CASE WHEN referrer_name LIKE '%openaire%' THEN 1 ELSE 0 END) AS openaire_referrer, extract('year' from timestamp::date) ||'/'|| LPAD(CAST(extract('month' from timestamp::date) AS VARCHAR), 2, '0') AS month, source FROM piwiklog where action='download' and (source_item_type='oaItem' or source_item_type='repItem') group by id, month, source order by source, id, month;";
+        stmt.executeUpdate(sql);
+
+        //sql = "SELECT 'OpenAIRE'::TEXT as source, d.id as repository_id, ro.id as result_id, month as date, max(downloads) AS count INTO downloads_stats FROM result_downloads_monthly p, datasource d, result_oids ro where p.source!='5' AND p.source=d.piwik_id and p.id=ro.orid group by repository_id, result_id, date ORDER BY repository_id, result_id, date;";
+        sql = "SELECT 'OpenAIRE'::TEXT as source, d.id as repository_id, ro.id as result_id, month as date, max(downloads) AS count, max(openaire_referrer) AS openaire INTO downloads_stats FROM result_downloads_monthly p, datasource d, result_oids ro where p.source!='5' AND p.source=d.piwik_id and p.id=ro.orid group by repository_id, result_id, date ORDER BY repository_id, result_id, date;";
+        stmt.executeUpdate(sql);
+
+        sql = "DROP VIEW IF EXISTS result_downloads_monthly;";
+        stmt.executeUpdate(sql);
+
+        stmt.close();
+        conn.commit();
+        conn.close();
+    }
+
+    public void finalizeStats() throws Exception {
+        if (conn.isClosed())
+            connectDB();
+
+        stmt = conn.createStatement();
+        conn.setAutoCommit(false);
+
+        Calendar startCalendar = Calendar.getInstance();
+        startCalendar.setTime(new SimpleDateFormat("yyyy-MM-dd").parse("2016-01-01"));
+        Calendar endCalendar = Calendar.getInstance();
+        int diffYear = endCalendar.get(Calendar.YEAR) - startCalendar.get(Calendar.YEAR);
+        int diffMonth = diffYear * 12 + endCalendar.get(Calendar.MONTH) - startCalendar.get(Calendar.MONTH);
+
+        String sql = "SELECT to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS full_date INTO full_dates FROM generate_series(0, " + diffMonth + ", 1) AS offs;";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX full_dates_full_date ON full_dates USING btree(full_date);";
+        stmt.executeUpdate(sql);
+
+
+        sql = "CREATE INDEX views_stats_source ON views_stats USING btree(source);";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX views_stats_repository_id ON views_stats USING btree(repository_id);";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX views_stats_result_id ON views_stats USING btree(result_id);";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX views_stats_date ON views_stats USING btree(date);";
+        stmt.executeUpdate(sql);
+
+
+        sql = "CREATE INDEX pageviews_stats_source ON pageviews_stats USING btree(source);";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX pageviews_stats_repository_id ON pageviews_stats USING btree(repository_id);";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX pageviews_stats_result_id ON pageviews_stats USING btree(result_id);";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX pageviews_stats_date ON pageviews_stats USING btree(date);";
+        stmt.executeUpdate(sql);
+
+
+        sql = "CREATE INDEX downloads_stats_source ON downloads_stats USING btree(source);";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX downloads_stats_repository_id ON downloads_stats USING btree(repository_id);";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX downloads_stats_result_id ON downloads_stats USING btree(result_id);";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX downloads_stats_date ON downloads_stats USING btree(date);";
+        stmt.executeUpdate(sql);
+
+
+        stmt.close();
+        conn.commit();
+        conn.close();
+    }
+
+    //views stats
+    private void viewsStatsOLD() throws Exception {
+        if (conn.isClosed())
+            connectDB();
+
+        stmt = conn.createStatement();
+        conn.setAutoCommit(false);
+
+        Calendar startCalendar = Calendar.getInstance();
+        startCalendar.setTime(new SimpleDateFormat("yyyy-MM-dd").parse("2016-01-01"));
+        Calendar endCalendar = Calendar.getInstance();
+        int diffYear = endCalendar.get(Calendar.YEAR) - startCalendar.get(Calendar.YEAR);
+        int diffMonth = diffYear * 12 + endCalendar.get(Calendar.MONTH) - startCalendar.get(Calendar.MONTH);
+
+        String sql = "CREATE OR REPLACE VIEW result_views_monthly as select entity_id AS id, COUNT(entity_id) as views, extract('year' from timestamp::date) ||'-'|| LPAD(CAST(extract('month' from timestamp::date) AS VARCHAR), 2, '0') ||'-01' AS month, source FROM piwiklog where action='action' and (source_item_type='oaItem' or source_item_type='repItem') group by id, month, source order by source, id, month;";
+        stmt.executeUpdate(sql);
+
+        sql = "SELECT d.id, d.new_date AS month, CASE when rdm.views IS NULL THEN 0 ELSE rdm.views END, d.source INTO result_views_sushi FROM (SELECT distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY-MM-DD') AS new_date, rdsm.source FROM generate_series(0, " + diffMonth + ", 1) AS offs, result_views_monthly rdsm) d LEFT JOIN (SELECT id, month, views, source FROM result_views_monthly) rdm ON d.new_date=rdm.month AND d.id=rdm.id AND d.source=rdm.source ORDER BY d.source, d.id, d.new_date;";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_views_sushi_id ON result_views_sushi USING btree (id);";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_views_sushi_month ON result_views_sushi USING btree (month);";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_views_sushi_source ON result_views_sushi USING btree (source);";
+        stmt.executeUpdate(sql);
+
+        sql = "SELECT roid.id, extract('year' from month::date) ||'/'|| LPAD(CAST(extract('month' from month::date) AS VARCHAR), 2, '0') as month, max(views) as views, source INTO result_views FROM result_views_sushi rvs, result_oids roid WHERE rvs.id=roid.orid GROUP BY source, roid.id, month ORDER BY source, roid.id, month;";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_views_id ON result_views USING btree (id);";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_views_month ON result_views USING btree (month);";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_views_source ON result_views USING btree (source);";
+        stmt.executeUpdate(sql);
+
+
+        sql = "DROP VIEW IF EXISTS result_views_monthly;";
+        stmt.executeUpdate(sql);
+
+        stmt.close();
+        conn.commit();
+        conn.close();
+    }
+
+    //downloads stats
+    private void downloadsStatsOLD() throws Exception {
+        if (conn.isClosed())
+            connectDB();
+
+        stmt = conn.createStatement();
+        conn.setAutoCommit(false);
+
+        Calendar startCalendar = Calendar.getInstance();
+        startCalendar.setTime(new SimpleDateFormat("yyyy-MM-dd").parse("2016-01-01"));
+        Calendar endCalendar = Calendar.getInstance();
+        int diffYear = endCalendar.get(Calendar.YEAR) - startCalendar.get(Calendar.YEAR);
+        int diffMonth = diffYear * 12 + endCalendar.get(Calendar.MONTH) - startCalendar.get(Calendar.MONTH);
+
+        String sql = "CREATE OR REPLACE VIEW result_downloads_monthly as select entity_id AS id, COUNT(entity_id) as downloads, extract('year' from timestamp::date) ||'-'|| LPAD(CAST(extract('month' from timestamp::date) AS VARCHAR), 2, '0') ||'-01' AS month, source FROM piwiklog where action='download' and (source_item_type='oaItem' or source_item_type='repItem') group by id, month, source order by source, id, month;";
+        stmt.executeUpdate(sql);
+
+        sql = "SELECT d.id, d.new_date AS month, CASE when rdm.downloads IS NULL THEN 0 ELSE rdm.downloads END, d.source INTO result_downloads_sushi FROM (SELECT distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY-MM-DD') AS new_date, rdsm.source FROM generate_series(0, " + diffMonth + ", 1) AS offs, result_downloads_monthly rdsm) d LEFT JOIN (SELECT id, month, downloads, source FROM result_downloads_monthly) rdm ON d.new_date=rdm.month AND d.id=rdm.id AND d.source=rdm.source ORDER BY d.source, d.id, d.new_date;";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_downloads_sushi_id ON result_downloads_sushi USING btree (id);";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_downloads_sushi_month ON result_downloads_sushi USING btree (month);";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_downloads_sushi_source ON result_downloads_sushi USING btree (source);";
+        stmt.executeUpdate(sql);
+
+        sql = "SELECT roid.id, extract('year' from month::date) ||'/'|| LPAD(CAST(extract('month' from month::date) AS VARCHAR), 2, '0') as month, downloads, source INTO result_downloads FROM result_downloads_sushi rvs, result_oids roid WHERE rvs.id=roid.orid ORDER BY source, roid.id, month;";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_downloads_id ON result_downloads USING btree (id);";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_downloads_month ON result_downloads USING btree (month);";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_downloads_source ON result_downloads USING btree (source);";
+        stmt.executeUpdate(sql);
+
+
+        sql = "DROP VIEW IF EXISTS result_downloads_monthly;";
+        stmt.executeUpdate(sql);
+
+        stmt.close();
+        conn.commit();
+        conn.close();
+    }
+
+    //Create repository Views statistics
+    private void repositoryViewsStats() throws Exception {
+        if (conn.isClosed())
+            connectDB();
+
+        stmt = conn.createStatement();
+        conn.setAutoCommit(false);
+
+        String sql = "SELECT entity_id AS id , COUNT(entity_id) AS number_of_views, timestamp::date AS date, source INTO repo_view_stats FROM piwiklog WHERE source!='5' AND action=\'action\' AND source_item_type=\'repItem\' GROUP BY entity_id, date, source ORDER BY entity_id, date ASC, COUNT(entity_id) DESC;";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX repo_view_stats_id ON repo_view_stats USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX repo_view_stats_date ON repo_view_stats USING btree(date)";
+        stmt.executeUpdate(sql);
+
+        sql = "SELECT roid.id, sum(number_of_views), extract('year' from date) ||'/'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') AS month, source INTO repo_view_stats_monthly_clean FROM repo_view_stats rvs, result_oids roid where rvs.id=roid.orid group by roid.id, month, source;";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX repo_view_stats_monthly_clean_id ON repo_view_stats_monthly_clean USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX repo_view_stats_monthly_clean_month ON repo_view_stats_monthly_clean USING btree(month)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX repo_view_stats_monthly_clean_source ON repo_view_stats_monthly_clean USING btree(source)";
+        stmt.executeUpdate(sql);
+
+        Calendar startCalendar = Calendar.getInstance();
+        startCalendar.setTime(new SimpleDateFormat("yyyy-MM-dd").parse("2016-01-01"));
+        Calendar endCalendar = Calendar.getInstance();
+        int diffYear = endCalendar.get(Calendar.YEAR) - startCalendar.get(Calendar.YEAR);
+        int diffMonth = diffYear * 12 + endCalendar.get(Calendar.MONTH) - startCalendar.get(Calendar.MONTH);
+
+        //sql="CREATE OR REPLACE view repo_view_stats_monthly AS select d.id, d.new_date AS month, case when rdm.sum is null then 0 else rdm.sum end, d.source from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS new_date, rdsm.source from generate_series(0, " + diffMonth +", 1) AS offs, repo_view_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum, source from repo_view_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id and d.source=rdm.source order by d.id, d.new_date";
+        sql = "select d.id, d.new_date AS month, case when rdm.sum is null then 0 else rdm.sum end, d.source INTO repo_view_stats_monthly from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS new_date, rdsm.source from generate_series(0, " + diffMonth + ", 1) AS offs, repo_view_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum, source from repo_view_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id and d.source=rdm.source order by d.id, d.new_date";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX repo_view_stats_monthly_id ON repo_view_stats_monthly USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX repo_view_stats_monthly_month ON repo_view_stats_monthly USING btree(month)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX repo_view_stats_monthly_source ON repo_view_stats_monthly USING btree(source)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE OR REPLACE view repo_view_stats_monthly_sushi AS SELECT id, sum(number_of_views), extract('year' from date) ||'-'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') ||'-01' AS month, source FROM repo_view_stats group by id, month, source;";
+        stmt.executeUpdate(sql);
+
+        stmt.close();
+        conn.commit();
+        conn.close();
+    }
+
+    //Create repository downloads statistics
+    private void repositoryDownloadsStats() throws Exception {
+        if (conn.isClosed())
+            connectDB();
+
+        stmt = conn.createStatement();
+        conn.setAutoCommit(false);
+
+        String sql = "SELECT entity_id AS id, COUNT(entity_id) AS number_of_downloads, timestamp::date AS date, source INTO repo_download_stats FROM piwiklog WHERE source!='5' AND action=\'download\' AND source_item_type=\'repItem\' GROUP BY entity_id, date, source ORDER BY entity_id, date ASC, COUNT(entity_id) DESC;";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX repo_download_stats_id ON repo_download_stats USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX repo_download_stats_date ON repo_download_stats USING btree(date)";
+        stmt.executeUpdate(sql);
+
+        sql = "SELECT roid.id, sum(number_of_downloads), extract('year' from date) ||'/'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') AS month, source INTO repo_download_stats_monthly_clean FROM repo_download_stats rvs, result_oids roid WHERE rvs.id=roid.orid GROUP BY roid.id, month, source;";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX repo_download_stats_monthly_clean_id ON repo_download_stats_monthly_clean USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX repo_download_stats_monthly_clean_month ON repo_download_stats_monthly_clean USING btree(month)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX repo_download_stats_monthly_clean_source ON repo_download_stats_monthly_clean USING btree(source)";
+        stmt.executeUpdate(sql);
+
+        Calendar startCalendar = Calendar.getInstance();
+        startCalendar.setTime(new SimpleDateFormat("yyyy-MM-dd").parse("2016-01-01"));
+        Calendar endCalendar = Calendar.getInstance();
+        int diffYear = endCalendar.get(Calendar.YEAR) - startCalendar.get(Calendar.YEAR);
+        int diffMonth = diffYear * 12 + endCalendar.get(Calendar.MONTH) - startCalendar.get(Calendar.MONTH);
+
+        //sql="CREATE OR REPLACE view repo_download_stats_monthly AS select d.id, d.new_date AS month, case when rdm.sum is null then 0 else rdm.sum end, d.source from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS new_date, rdsm.source from generate_series(0, " + diffMonth +", 1) AS offs, repo_download_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum, source from repo_download_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id and d.source=rdm.source order by d.id, d.new_date";
+        sql = "select d.id, d.new_date AS month, case when rdm.sum is null then 0 else rdm.sum end, d.source INTO repo_download_stats_monthly from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS new_date, rdsm.source from generate_series(0, " + diffMonth + ", 1) AS offs, repo_download_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum, source from repo_download_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id and d.source=rdm.source order by d.id, d.new_date";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX repo_download_stats_monthly_id ON repo_download_stats_monthly USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX repo_download_stats_monthly_month ON repo_download_stats_monthly USING btree(month)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX repo_download_stats_monthly_source ON repo_download_stats_monthly USING btree(source)";
+        stmt.executeUpdate(sql);
+
+
+        sql = "CREATE OR REPLACE view repo_download_stats_monthly_sushi AS SELECT id, sum(number_of_downloads), extract('year' from date) ||'-'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') ||'-01' AS month, source FROM repo_download_stats group by id, month, source;";
+        stmt.executeUpdate(sql);
+
+        stmt.close();
+        conn.commit();
+        conn.close();
+    }
+
+    // Import OPENAIRE Logs to DB
+    private void processPortalLog() throws Exception {
+
+        if (conn.isClosed())
+            connectDB();
+
+        stmt = conn.createStatement();
+        conn.setAutoCommit(false);
+
+        ArrayList<String> jsonFiles = listHdfsDir(logPath + "portallog");
+        PreparedStatement prepStatem = conn.prepareStatement("INSERT INTO process_portal_log (source, id_visit, country, action, url, entity_id, source_item_type, timestamp, referrer_name, agent) VALUES (?,?,?,?,?,?,?,?,?,?)");
+        int batch_size = 0;
+        JSONParser parser = new JSONParser();
+        for (String jsonFile : jsonFiles) {
+            JSONArray jsonArray = (JSONArray) parser.parse(readHDFSFile(jsonFile));
+
+            for (Object aJsonArray : jsonArray) {
+                JSONObject jsonObjectRow = (JSONObject) aJsonArray;
+                int idSite = Integer.parseInt(jsonObjectRow.get("idSite").toString());
+                String idVisit = jsonObjectRow.get("idVisit").toString();
+                String country = jsonObjectRow.get("country").toString();
+                String referrerName = jsonObjectRow.get("referrerName").toString();
+                String agent = jsonObjectRow.get("browser").toString();
+
+                JSONArray actionDetails = (JSONArray) jsonObjectRow.get(("actionDetails"));
+                for (Object actionDetail : actionDetails) {
+                    JSONObject actionDetailsObj = (JSONObject) actionDetail;
+
+                    SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+                    simpleDateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));
+                    Timestamp timestamp = new Timestamp(Long.parseLong(actionDetailsObj.get("timestamp").toString()) * 1000);
+
+                    String action = actionDetailsObj.get("type").toString();
+                    String url = actionDetailsObj.get("url").toString();
+
+                    String entityID = processPortalURL(url);
+                    String sourceItemType = "";
+
+                    if (entityID.indexOf("|") > 0) {
+                        sourceItemType = entityID.substring(0, entityID.indexOf("|"));
+                        entityID = entityID.substring(entityID.indexOf("|") + 1);
+                    }
+
+                    prepStatem.setInt(1, idSite);
+                    prepStatem.setString(2, idVisit);
+                    prepStatem.setString(3, country);
+                    prepStatem.setString(4, action);
+                    prepStatem.setString(5, url);
+                    prepStatem.setString(6, entityID);
+                    prepStatem.setString(7, sourceItemType);
+                    prepStatem.setString(8, simpleDateFormat.format(timestamp));
+                    prepStatem.setString(9, referrerName);
+                    prepStatem.setString(10, agent);
+
+                    prepStatem.addBatch();
+                    batch_size++;
+                    if (batch_size == 10000) {
+                        prepStatem.executeBatch();
+                        conn.commit();
+                        batch_size = 0;
+                    }
+                }
+            }
+        }
+        prepStatem.executeBatch();
+        conn.commit();
+
+        stmt.close();
+        conn.close();
+    }
+
+    private void portalStats() throws Exception {
+        if (conn.isClosed())
+            connectDB();
+
+        stmt = conn.createStatement();
+        conn.setAutoCommit(false);
+
+        String sql = "INSERT INTO piwiklog SELECT DISTINCT source, id_visit, country, action, url, roid.orid, \'oaItem\', timestamp, referrer_name, agent FROM process_portal_log, public.result_oids roid WHERE entity_id IS NOT null AND entity_id=roid.id AND roid.orid IS NOT null;";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "INSERT INTO piwiklog SELECT DISTINCT source, id_visit, country, action, url, roid.orid, \'datasource\', timestamp, referrer_name, agent FROM process_portal_log, public.datasource_oids roid WHERE entity_id IS NOT null AND entity_id=roid.id AND roid.orid IS NOT null;";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "INSERT INTO piwiklog SELECT DISTINCT source, id_visit, country, action, url, roid.orid, \'organization\', timestamp, referrer_name, agent FROM process_portal_log, public.organization_oids roid WHERE entity_id IS NOT null AND entity_id=roid.id AND roid.orid IS NOT null;";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "INSERT INTO piwiklog SELECT DISTINCT source, id_visit, country, action, url, roid.orid, \'project\', timestamp, referrer_name, agent FROM process_portal_log, public.project_oids roid WHERE entity_id IS NOT null AND entity_id=roid.id AND roid.orid IS NOT null;";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "DROP TABLE process_portal_log;";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        conn.close();
+    }
+
+    private void cleanOAI() throws Exception {
+        if (conn.isClosed())
+            connectDB();
+
+        conn.setAutoCommit(false);
+
+        stmt = conn.createStatement();
+        String sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorio.chlc.min-saude.pt/','oai:repositorio.chlc.min-saude.pt:') WHERE entity_id LIKE 'oai:repositorio.chlc.min-saude.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorio.hospitaldebraga.pt/','oai:repositorio.hospitaldebraga.pt:') WHERE entity_id LIKE 'oai:repositorio.hospitaldebraga.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorio.ipl.pt/','oai:repositorio.ipl.pt:') WHERE entity_id LIKE 'oai:repositorio.ipl.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:bibliotecadigital.ipb.pt/','oai:bibliotecadigital.ipb.pt:') WHERE entity_id LIKE 'oai:bibliotecadigital.ipb.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorio.ismai.pt/','oai:repositorio.ismai.pt:') WHERE entity_id LIKE 'oai:repositorio.ismai.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorioaberto.uab.pt/','oai:repositorioaberto.uab.pt:') WHERE entity_id LIKE 'oai:repositorioaberto.uab.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorio.uac.pt/','oai:repositorio.uac.pt:') WHERE entity_id LIKE 'oai:repositorio.uac.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorio.insa.pt/','oai:repositorio.insa.pt:') WHERE entity_id LIKE 'oai:repositorio.insa.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorio.ipcb.pt/','oai:repositorio.ipcb.pt:') WHERE entity_id LIKE 'oai:repositorio.ipcb.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorio.ispa.pt/','oai:repositorio.ispa.pt:') WHERE entity_id LIKE 'oai:repositorio.ispa.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorio.chporto.pt/','oai:repositorio.chporto.pt:') WHERE entity_id LIKE 'oai:repositorio.chporto.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorio.ucp.pt/','oai:repositorio.ucp.pt:') WHERE entity_id LIKE 'oai:repositorio.ucp.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:rihuc.huc.min-saude.pt/','oai:rihuc.huc.min-saude.pt:') WHERE entity_id LIKE 'oai:rihuc.huc.min-saude.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorio.ipv.pt/','oai:repositorio.ipv.pt:') WHERE entity_id LIKE 'oai:repositorio.ipv.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:www.repository.utl.pt/','oai:www.repository.utl.pt:') WHERE entity_id LIKE 'oai:www.repository.utl.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:run.unl.pt/','oai:run.unl.pt:') WHERE entity_id LIKE 'oai:run.unl.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:sapientia.ualg.pt/','oai:sapientia.ualg.pt:') WHERE entity_id LIKE 'oai:sapientia.ualg.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorio.ipsantarem.pt/','oai:repositorio.ipsantarem.pt:') WHERE entity_id LIKE 'oai:repositorio.ipsantarem.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:arca.igc.gulbenkian.pt/','oai:arca.igc.gulbenkian.pt:') WHERE entity_id LIKE 'oai:arca.igc.gulbenkian.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:ubibliorum.ubi.pt/','oai:ubibliorum.ubi.pt:') WHERE entity_id LIKE 'oai:ubibliorum.ubi.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:digituma.uma.pt/','oai:digituma.uma.pt:') WHERE entity_id LIKE 'oai:digituma.uma.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorio.ul.pt/','oai:repositorio.ul.pt:') WHERE entity_id LIKE 'oai:repositorio.ul.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorio.hff.min-saude.pt/','oai:repositorio.hff.min-saude.pt:') WHERE entity_id LIKE 'oai:repositorio.hff.min-saude.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorium.sdum.uminho.pt/','oai:repositorium.sdum.uminho.pt:') WHERE entity_id LIKE 'oai:repositorium.sdum.uminho.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:recipp.ipp.pt/','oai:recipp.ipp.pt:') WHERE entity_id LIKE 'oai:recipp.ipp.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:bdigital.ufp.pt/','oai:bdigital.ufp.pt:') WHERE entity_id LIKE 'oai:bdigital.ufp.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorio.lneg.pt/','oai:repositorio.lneg.pt:') WHERE entity_id LIKE 'oai:repositorio.lneg.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:iconline.ipleiria.pt/','oai:iconline.ipleiria.pt:') WHERE entity_id LIKE 'oai:iconline.ipleiria.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        stmt = conn.createStatement();
+        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:comum.rcaap.pt/','oai:comum.rcaap.pt:') WHERE entity_id LIKE 'oai:comum.rcaap.pt/%';";
+        stmt.executeUpdate(sql);
+        stmt.close();
+        conn.commit();
+
+        conn.close();
+    }
+
+    //Create OpenAIRE's portal datasource statistics
+    private void dataSourceStats() throws Exception {
+        if (conn.isClosed())
+            connectDB();
+
+        stmt = conn.createStatement();
+        conn.setAutoCommit(false);
+
+        String sql = "SELECT  orgid AS id, max(viewcount) AS number_of_views, date INTO datasource_stats FROM (select entity_id, ooid.id AS orgid, count(entity_id) viewcount, timestamp::date as date FROM piwiklog, datasource_oids ooid WHERE entity_id=ooid.orid AND source_item_type='datasource' GROUP BY entity_id, ooid.id, date) AS temp GROUP BY orgid, date ORDER BY orgid, date ASC, max(viewcount) DESC;";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX datasource_stats_id ON datasource_stats USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX datasource_stats_date ON datasource_stats USING btree(date)";
+        stmt.executeUpdate(sql);
+
+        sql = "SELECT id, sum(number_of_views), extract('year' from date) ||'/'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') AS month INTO datasource_stats_monthly_clean FROM datasource_stats GROUP BY id, month;";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX datasource_stats_monthly_clean_id ON datasource_stats_monthly_clean USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX datasource_stats_monthly_clean_month ON datasource_stats_monthly_clean USING btree(month)";
+        stmt.executeUpdate(sql);
+
+        Calendar startCalendar = Calendar.getInstance();
+        startCalendar.setTime(new SimpleDateFormat("yyyy-MM-dd").parse("2016-01-01"));
+        Calendar endCalendar = Calendar.getInstance();
+        int diffYear = endCalendar.get(Calendar.YEAR) - startCalendar.get(Calendar.YEAR);
+        int diffMonth = diffYear * 12 + endCalendar.get(Calendar.MONTH) - startCalendar.get(Calendar.MONTH);
+
+        //sql="CREATE OR REPLACE view datasource_stats_monthly AS select d.id, d.new_date as month, case when rdm.sum is null then 0 else rdm.sum end from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY-MM-DD') AS new_date from generate_series(0, " + diffMonth +", 1) AS offs, datasource_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum from datasource_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id order by d.id, d.new_date";
+        sql = "select d.id, d.new_date AS month, case when rdm.sum is null then 0 else rdm.sum end INTO datasource_stats_monthly from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY-MM-DD') AS new_date from generate_series(0, " + diffMonth + ", 1) AS offs, datasource_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum from datasource_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id order by d.id, d.new_date";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX datasource_stats_monthly_id ON datasource_stats_monthly USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX datasource_stats_monthly_month ON datasource_stats_monthly USING btree(month)";
+        stmt.executeUpdate(sql);
+
+        stmt.close();
+        conn.commit();
+        conn.close();
+    }
+
+    //Create OpenAIRE's portal results statistics
+    private void resultStats() throws Exception {
+        if (conn.isClosed())
+            connectDB();
+
+        stmt = conn.createStatement();
+        conn.setAutoCommit(false);
+
+        String sql = "SELECT orgid AS id, max(viewcount) AS number_of_views, date INTO result_stats FROM (SELECT entity_id, ooid.id AS orgid, count(entity_id) viewcount, timestamp::date as date FROM piwiklog, result_oids ooid WHERE entity_id=ooid.orid AND source_item_type='oaItem' GROUP BY entity_id, ooid.id, date) AS temp GROUP BY orgid, date ORDER BY orgid, date ASC, max(viewcount) DESC;";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_stats_id ON result_stats USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_stats_date ON result_stats USING btree(date)";
+        stmt.executeUpdate(sql);
+
+        sql = "SELECT id, sum(number_of_views), extract('year' from date) ||'/'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') AS month INTO result_stats_monthly_clean FROM result_stats group by id, month;";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_stats_monthly_clean_id ON result_stats_monthly_clean USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_stats_monthly_clean_month ON result_stats_monthly_clean USING btree(month)";
+        stmt.executeUpdate(sql);
+
+        Calendar startCalendar = Calendar.getInstance();
+        startCalendar.setTime(new SimpleDateFormat("yyyy-MM-dd").parse("2016-01-01"));
+        Calendar endCalendar = Calendar.getInstance();
+        int diffYear = endCalendar.get(Calendar.YEAR) - startCalendar.get(Calendar.YEAR);
+        int diffMonth = diffYear * 12 + endCalendar.get(Calendar.MONTH) - startCalendar.get(Calendar.MONTH);
+
+        //sql="CREATE OR REPLACE view result_stats_monthly AS select d.id, d.new_date as month, case when rdm.sum is null then 0 else rdm.sum end from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS new_date from generate_series(0, " + diffMonth +", 1) AS offs, result_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum from result_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id order by d.id, d.new_date";
+        sql = "select d.id, d.new_date as month, case when rdm.sum is null then 0 else rdm.sum end INTO result_stats_monthly from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS new_date from generate_series(0, " + diffMonth + ", 1) AS offs, result_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum from result_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id order by d.id, d.new_date";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_stats_monthly_id ON result_stats_monthly USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX result_stats_monthly_month ON result_stats_monthly USING btree(month)";
+        stmt.executeUpdate(sql);
+
+        stmt.close();
+        conn.commit();
+        conn.close();
+    }
+
+    //Create OpenAIRE's portal organization statistics
+    private void organizationsStats() throws Exception {
+        if (conn.isClosed())
+            connectDB();
+
+        stmt = conn.createStatement();
+        conn.setAutoCommit(false);
+
+        String sql = "SELECT orgid AS id, max(viewcount) AS number_of_views, date INTO organization_stats FROM (SELECT entity_id, ooid.id AS orgid, count(entity_id) viewcount, timestamp::date AS date FROM piwiklog, organization_oids ooid WHERE entity_id=ooid.orid AND source_item_type='organization' GROUP BY entity_id, ooid.id, date) AS temp GROUP BY orgid, date ORDER BY orgid, date ASC, max(viewcount) DESC;";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX organization_stats_id ON organization_stats USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX organization_stats_date ON organization_stats USING btree(date)";
+        stmt.executeUpdate(sql);
+
+        sql = "SELECT id, sum(number_of_views), extract('year' from date) ||'/'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') AS month INTO organization_stats_monthly_clean FROM organization_stats group by id, month;";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX organization_stats_monthly_clean_id ON organization_stats_monthly_clean USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX organization_stats_monthly_clean_month ON organization_stats_monthly_clean USING btree(month)";
+        stmt.executeUpdate(sql);
+
+        Calendar startCalendar = Calendar.getInstance();
+        startCalendar.setTime(new SimpleDateFormat("yyyy-MM-dd").parse("2016-01-01"));
+        Calendar endCalendar = Calendar.getInstance();
+        int diffYear = endCalendar.get(Calendar.YEAR) - startCalendar.get(Calendar.YEAR);
+        int diffMonth = diffYear * 12 + endCalendar.get(Calendar.MONTH) - startCalendar.get(Calendar.MONTH);
+
+        //sql="CREATE OR REPLACE view organization_stats_monthly AS select d.id, d.new_date as month, case when rdm.sum is null then 0 else rdm.sum end from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY-MM-DD') AS new_date from generate_series(0, " + diffMonth +", 1) AS offs, organization_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum from organization_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id order by d.id, d.new_date";
+        sql = "select d.id, d.new_date as month, case when rdm.sum is null then 0 else rdm.sum end INTO organization_stats_monthly from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY-MM-DD') AS new_date from generate_series(0, " + diffMonth + ", 1) AS offs, organization_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum from organization_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id order by d.id, d.new_date";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX organization_stats_monthly_id ON organization_stats_monthly USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX organization_stats_monthly_month ON organization_stats_monthly USING btree(month)";
+        stmt.executeUpdate(sql);
+
+        stmt.close();
+        conn.commit();
+        conn.close();
+    }
+
+    //Create OpenAIRE's portal projects statistics
+    private void projectsStats() throws Exception {
+        if (conn.isClosed())
+            connectDB();
+
+        stmt = conn.createStatement();
+        conn.setAutoCommit(false);
+
+        String sql = "SELECT orgid AS id, max(viewcount) AS number_of_views, date INTO project_stats FROM (SELECT entity_id, ooid.id AS orgid, count(entity_id) viewcount, timestamp::date AS date FROM piwiklog, project_oids ooid WHERE entity_id=ooid.orid AND source_item_type='project' GROUP BY entity_id, ooid.id, date) AS temp GROUP BY orgid, date ORDER BY orgid, date ASC, max(viewcount) DESC;";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX project_stats_id ON project_stats USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX project_stats_date ON project_stats USING btree(date)";
+        stmt.executeUpdate(sql);
+
+        sql = "SELECT id, sum(number_of_views), extract('year' from date) ||'/'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') AS month INTO project_stats_monthly_clean FROM project_stats group by id, month;";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX project_stats_monthly_clean_id ON project_stats_monthly_clean USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX project_stats_monthly_clean_month ON project_stats_monthly_clean USING btree(month)";
+        stmt.executeUpdate(sql);
+
+        Calendar startCalendar = Calendar.getInstance();
+        startCalendar.setTime(new SimpleDateFormat("yyyy-MM-dd").parse("2016-01-01"));
+        Calendar endCalendar = Calendar.getInstance();
+        int diffYear = endCalendar.get(Calendar.YEAR) - startCalendar.get(Calendar.YEAR);
+        int diffMonth = diffYear * 12 + endCalendar.get(Calendar.MONTH) - startCalendar.get(Calendar.MONTH);
+
+        //sql="CREATE OR REPLACE view project_stats_monthly AS select d.id, d.new_date as month, case when rdm.sum is null then 0 else rdm.sum end from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS new_date from generate_series(0, " + diffMonth +", 1) AS offs, project_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum from project_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id order by d.id, d.new_date";
+        sql = "select d.id, d.new_date as month, case when rdm.sum is null then 0 else rdm.sum end INTO project_stats_monthly from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS new_date from generate_series(0, " + diffMonth + ", 1) AS offs, project_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum from project_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id order by d.id, d.new_date";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX project_stats_monthly_id ON project_stats_monthly USING btree (id)";
+        stmt.executeUpdate(sql);
+
+        sql = "CREATE INDEX project_stats_monthly_month ON project_stats_monthly USING btree(month)";
+        stmt.executeUpdate(sql);
+
+        stmt.close();
+        conn.commit();
+        conn.close();
+    }
+
+    private String processPortalURL(String url) {
+
+        if (url.indexOf("openaire.eu") > 0) {
+            try {
+                url = URLDecoder.decode(url, "UTF-8");
+            } catch (Exception e) {
+                log.info(url);
+            }
+            if (url.indexOf("datasourceId=") > 0 && url.substring(url.indexOf("datasourceId=") + 13).length() >= 46) {
+                url = "datasource|" + url.substring(url.indexOf("datasourceId=") + 13, url.indexOf("datasourceId=") + 59);
+            } else if (url.indexOf("datasource=") > 0 && url.substring(url.indexOf("datasource=") + 11).length() >= 46) {
+                url = "datasource|" + url.substring(url.indexOf("datasource=") + 11, url.indexOf("datasource=") + 57);
+            } else if (url.indexOf("datasourceFilter=") > 0 && url.substring(url.indexOf("datasourceFilter=") + 17).length() >= 46) {
+                url = "datasource|" + url.substring(url.indexOf("datasourceFilter=") + 17, url.indexOf("datasourceFilter=") + 63);
+            } else if (url.indexOf("articleId=") > 0 && url.substring(url.indexOf("articleId=") + 10).length() >= 46) {
+                url = "result|" + url.substring(url.indexOf("articleId=") + 10, url.indexOf("articleId=") + 56);
+            } else if (url.indexOf("datasetId=") > 0 && url.substring(url.indexOf("datasetId=") + 10).length() >= 46) {
+                url = "result|" + url.substring(url.indexOf("datasetId=") + 10, url.indexOf("datasetId=") + 56);
+            } else if (url.indexOf("projectId=") > 0 && url.substring(url.indexOf("projectId=") + 10).length() >= 46 && !url.contains("oai:dnet:corda")) {
+                url = "project|" + url.substring(url.indexOf("projectId=") + 10, url.indexOf("projectId=") + 56);
+            } else if (url.indexOf("organizationId=") > 0 && url.substring(url.indexOf("organizationId=") + 15).length() >= 46) {
+                url = "organization|" + url.substring(url.indexOf("organizationId=") + 15, url.indexOf("organizationId=") + 61);
+            } else {
+                url = "";
+            }
+        } else {
+            url = "";
+        }
+
+        return url;
+    }
+
+    private ArrayList<String> listHdfsDir(String dir) throws Exception {
+
+        FileSystem hdfs = FileSystem.get(new Configuration());
+        RemoteIterator<LocatedFileStatus> Files;
+        ArrayList<String> fileNames = new ArrayList<>();
+
+        try {
+            Path exportPath = new Path(hdfs.getUri() + dir);
+            Files = hdfs.listFiles(exportPath, false);
+            while (Files.hasNext()) {
+                String fileName = Files.next().getPath().toString();
+                //log.info("Found hdfs file " + fileName);
+                fileNames.add(fileName);
+            }
+            //hdfs.close();
+        } catch (Exception e) {
+            log.error("HDFS file path with exported data does not exist : " + new Path(hdfs.getUri() + logPath));
+            throw new Exception("HDFS file path with exported data does not exist :   " + logPath, e);
+        }
+
+        return fileNames;
+    }
+
+    private String readHDFSFile(String filename) throws Exception {
+        String result;
+        try {
+
+            FileSystem fs = FileSystem.get(new Configuration());
+            //log.info("reading file : " + filename);
+
+            BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(new Path(filename))));
+
+            StringBuilder sb = new StringBuilder();
+            String line = br.readLine();
+
+            while (line != null) {
+                if (!line.equals("[]")) {
+                    sb.append(line);
+                }
+                //sb.append(line);
+                line = br.readLine();
+            }
+            result = sb.toString().replace("][{\"idSite\"", ",{\"idSite\"");
+            if (result.equals("")) {
+                result = "[]";
+            }
+
+            //fs.close();
+        } catch (Exception e) {
+            log.error(e);
+            throw new Exception(e);
+        }
+
+        return result;
+    }
+}
Index: modules/dnet-openaire-usage-stats-export/trunk/src/main/java/eu/dnetlib/usagestats/export/UsageStatsExporter.java
===================================================================
--- modules/dnet-openaire-usage-stats-export/trunk/src/main/java/eu/dnetlib/usagestats/export/UsageStatsExporter.java	(nonexistent)
+++ modules/dnet-openaire-usage-stats-export/trunk/src/main/java/eu/dnetlib/usagestats/export/UsageStatsExporter.java	(revision 47499)
@@ -0,0 +1,231 @@
+package eu.dnetlib.usagestats.export;
+
+import org.apache.log4j.Logger;
+
+public class UsageStatsExporter {
+
+    private String username;
+    private String password;
+    private String httpProtocol;
+    private String piwikUrl;
+    private String siteId;
+    private String startDate;
+    private String finalDate;
+    private String logsPath;
+    private String filterOffset;
+
+
+    private String tokenAuth;
+    private String irusBaseURL;
+    private String irusReportType;
+    private String irusRelease;
+    private String irusRequestorID;
+    private String irusRepositoryIdentifier;
+    private String irusPeriodGranularity;
+    private String irusPeriod;
+
+
+    private String dbUrl;
+    private String dbSchema;
+    private String dbUsername;
+    private String dbPassword;
+
+    private Logger log = Logger.getLogger(this.getClass());
+
+    public UsageStatsExporter() {}
+
+    public void export() throws Exception {
+
+        PiwikDownloadLogs piwd = new PiwikDownloadLogs(username, password, tokenAuth, httpProtocol, piwikUrl, startDate, logsPath, dbUrl, dbUsername, dbPassword);
+        piwd.getPiwikLogs();
+
+        /*
+         *  Create DB tables, insert/update statistics
+         */
+
+        PiwikStatsDB piwikstats = new PiwikStatsDB(dbUrl, dbUsername, dbPassword, logsPath);
+        piwikstats.processLogs();
+
+        //Create Usage Stats tables from Downloaded Logs
+        piwikstats.usageStats();
+
+
+        SushiStats sushiStats = new SushiStats(dbUrl, dbUsername, dbPassword);
+        sushiStats.processIrusRRReport();
+        sushiStats.sushiStats();
+
+
+        //finalize stats
+        piwikstats.finalizeStats();
+
+    }
+
+    public String getUsername() {
+        return username;
+    }
+
+    public void setUsername(String username) {
+        this.username = username;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
+    public String getHttpProtocol() {
+        return httpProtocol;
+    }
+
+    public void setHttpProtocol(String httpProtocol) {
+        this.httpProtocol = httpProtocol;
+    }
+
+    public String getPiwikUrl() {
+        return piwikUrl;
+    }
+
+    public void setPiwikUrl(String piwikUrl) {
+        this.piwikUrl = piwikUrl;
+    }
+
+    public String getSiteId() {
+        return siteId;
+    }
+
+    public void setSiteId(String siteId) {
+        this.siteId = siteId;
+    }
+
+    public String getStartDate() {
+        return startDate;
+    }
+
+    public void setStartDate(String startDate) {
+        this.startDate = startDate;
+    }
+
+    public String getFinalDate() {
+        return finalDate;
+    }
+
+    public void setFinalDate(String finalDate) {
+        this.finalDate = finalDate;
+    }
+
+    public String getLogsPath() {
+        return logsPath;
+    }
+
+    public void setLogsPath(String logsPath) {
+        this.logsPath = logsPath;
+    }
+
+    public String getFilterOffset() {
+        return filterOffset;
+    }
+
+    public void setFilterOffset(String filterOffset) {
+        this.filterOffset = filterOffset;
+    }
+
+    public String getDbUrl() {
+        return dbUrl;
+    }
+
+    public void setDbUrl(String dbUrl) {
+        this.dbUrl = dbUrl;
+    }
+
+    public String getDbSchema() {
+        return dbSchema;
+    }
+
+    public void setDbSchema(String dbSchema) {
+        this.dbSchema = dbSchema;
+    }
+
+
+    public String getDbUsername() {
+        return dbUsername;
+    }
+
+    public void setDbUsername(String dbUsername) {
+        this.dbUsername = dbUsername;
+    }
+
+    public String getDbPassword() {
+        return dbPassword;
+    }
+
+    public void setDbPassword(String dbPassword) {
+        this.dbPassword = dbPassword;
+    }
+
+    public String getTokenAuth() {
+        return tokenAuth;
+    }
+
+    public void setTokenAuth(String tokenAuth) {
+        this.tokenAuth = tokenAuth;
+    }
+
+    public String getIrusBaseURL() {
+        return irusBaseURL;
+    }
+
+    public void setIrusBaseURL(String irusBaseURL) {
+        this.irusBaseURL = irusBaseURL;
+    }
+
+    public String getIrusReportType() {
+        return irusReportType;
+    }
+
+    public void setIrusReportType(String irusReportType) {
+        this.irusReportType = irusReportType;
+    }
+
+    public String getIrusRelease() {
+        return irusRelease;
+    }
+
+    public void setIrusRelease(String irusRelease) {
+        this.irusRelease = irusRelease;
+    }
+
+    public String getIrusRequestorID() {
+        return irusRequestorID;
+    }
+
+    public void setIrusRequestorID(String irusRequestorID) {
+        this.irusRequestorID = irusRequestorID;
+    }
+
+    public String getIrusRepositoryIdentifier() {
+        return irusRepositoryIdentifier;
+    }
+
+    public void setIrusRepositoryIdentifier(String irusRepositoryIdentifier) {
+        this.irusRepositoryIdentifier = irusRepositoryIdentifier;
+    }
+
+    public String getIrusPeriodGranularity() {
+        return irusPeriodGranularity;
+    }
+
+    public void setIrusPeriodGranularity(String irusPeriodGranularity) {
+        this.irusPeriodGranularity = irusPeriodGranularity;
+    }
+
+    public String getIrusPeriod() {
+        return irusPeriod;
+    }
+
+    public void setIrusPeriod(String irusPeriod) {
+        this.irusPeriod = irusPeriod;
+    }
+}
Index: modules/dnet-openaire-usage-stats-export/trunk/src/main/java/eu/dnetlib/usagestats/export/PiwikDownloadLogs.java
===================================================================
--- modules/dnet-openaire-usage-stats-export/trunk/src/main/java/eu/dnetlib/usagestats/export/PiwikDownloadLogs.java	(nonexistent)
+++ modules/dnet-openaire-usage-stats-export/trunk/src/main/java/eu/dnetlib/usagestats/export/PiwikDownloadLogs.java	(revision 47499)
@@ -0,0 +1,214 @@
+package eu.dnetlib.usagestats.export;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.log4j.Logger;
+
+import java.io.*;
+import java.net.URL;
+import java.net.URLConnection;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Calendar;
+
+public class PiwikDownloadLogs {
+
+    private final String piwikUsername;
+    private final String piwikPassword;
+    private final String httpProtocol;
+    private final String piwikUrl;
+    private final Date startDate;
+    private final String tokenAuth;
+    private final String logsPath;
+
+    private final String dbUrl;
+    private final String dbUserName;
+    private final String dbPassword;
+
+    /*
+       The Piwik's API method 
+    */
+    private final String APImethod = "?module=API&method=Live.getLastVisitsDetails";
+    private final String format = "&format=json";
+
+    private final Logger log = Logger.getLogger(this.getClass());
+
+
+    public PiwikDownloadLogs(String username, String password, String tokenAuth, String httpProtocol, String piwikURl, String sDate, String logsPath, String dbUrl, String dbUsername, String dbPassword) throws Exception{
+        this.piwikUsername = username;
+        this.piwikPassword = password;
+        this.httpProtocol = httpProtocol;
+        this.piwikUrl = piwikURl;
+
+        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
+        this.startDate = sdf.parse(sDate);
+
+        this.tokenAuth = tokenAuth;
+        this.logsPath = logsPath;
+        this.dbUrl = dbUrl;
+        this.dbUserName = dbUsername;
+        this.dbPassword = dbPassword;
+    }
+
+    private String getPiwikLogUrl(){
+        return httpProtocol + "://" + piwikUrl + "/";
+    }
+
+    private String getJson(String url,String username, String password) throws Exception {
+        //String cred=username+":"+password;
+        //String encoded = new sun.misc.BASE64Encoder().encode (cred.getBytes());
+        try {
+            URL website = new URL(url);
+            URLConnection connection = website.openConnection();
+
+            //connection.setRequestProperty ("Authorization", "Basic "+encoded);
+            StringBuilder response;
+            try (BufferedReader in = new BufferedReader(new InputStreamReader(connection.getInputStream()))) {
+                response = new StringBuilder();
+                String inputLine;
+                while ((inputLine = in.readLine()) != null) {
+                    response.append(inputLine);
+                    response.append("\n");
+                }
+            }
+            return response.toString();
+        }catch (Exception e){
+            log.error("Failed to get URL: " + e);
+            throw new Exception("Failed to get URL: " + e.toString(), e);
+        }
+    }
+
+    public void getPiwikLogs() throws Exception{
+        GetPortalLogs();
+        GetRepositoriesLogs();
+    }
+
+    private void GetPortalLogs() throws Exception{
+
+        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
+        Calendar start = Calendar.getInstance();
+        start.setTime(startDate);
+        Calendar end = Calendar.getInstance();
+        end.add(Calendar.DAY_OF_MONTH, -1);
+        //end.setTime(getFinalDate());
+
+        try{
+            log.info("downloading logs for site with piwik_id: 5");
+            Class.forName("org.postgresql.Driver");
+            Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword);
+            PreparedStatement st = conn.prepareStatement("SELECT max(timestamp) FROM public.piwiklog WHERE source='5';");
+            ResultSet rs_date = st.executeQuery();
+            while(rs_date.next()){
+                start.setTime(sdf.parse(rs_date.getString(1)));
+            }
+            rs_date.close();
+            conn.close();
+            for (Date date = start.getTime(); start.before(end); start.add(Calendar.DATE, 1), date = start.getTime()) {
+
+                String period="&period=day&date="+sdf.format(date);
+
+                String apiUrl=getPiwikLogUrl()+APImethod+"&idSite=5"+period+format+"&expanded=5&filter_limit=1000&token_auth="+tokenAuth;
+                String content = getJson(apiUrl,piwikUsername,piwikPassword);
+
+                //for (int i=1;i<10;i++){
+                int i = 1;
+                while(true) {
+                    String apiUrlnew=apiUrl+"&filter_offset="+i*1000;
+                    String contentNew = getJson(apiUrlnew,piwikUsername,piwikUsername);
+                    content += contentNew;
+                    i++;
+                    if(contentNew.equals("[]\n")){
+                        break;
+                    }
+                }
+                flushString(content, logsPath + "portallog/" + "5_Piwiklog"+sdf.format((date))+".json");
+            }
+        } catch (Exception e) {
+            log.error(e);
+            throw new Exception("Failed to get portal logs: " + e.toString(), e);
+        }
+    }
+
+    private void GetRepositoriesLogs() throws Exception{
+
+        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
+        Calendar start = Calendar.getInstance();
+        start.setTime(startDate);
+        Calendar end = Calendar.getInstance();
+        end.add(Calendar.DAY_OF_MONTH, -1);
+        //end.setTime(getFinalDate());
+
+        try{
+            Class.forName("org.postgresql.Driver");
+            Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword);
+            Statement statement = conn.createStatement();
+            ResultSet rs = statement.executeQuery("SELECT distinct piwik_id from shadow.datasource where piwik_id is not null and piwik_id!='5' order by piwik_id;");
+            while(rs.next()){
+                start.setTime(startDate);
+                int siteId = rs.getInt(1);
+                log.info("downloading logs for site with piwik_id: " + siteId);
+                PreparedStatement st = conn.prepareStatement("SELECT max(timestamp) FROM public.piwiklog WHERE source=?;");
+                st.setInt(1, siteId);
+                ResultSet rs_date = st.executeQuery();
+                while(rs_date.next()){
+                    //log.info("source: " + siteId + " - date: " + rs_date.getString(1));
+                    if(rs_date.getString(1) == null || rs_date.getString(1).equals("null") || rs_date.getString(1).equals("")) {
+                        start = Calendar.getInstance();
+                        start.add(Calendar.MONTH, -1);
+                    }
+                    else {
+                        start.setTime(sdf.parse(rs_date.getString(1)));
+                    }
+                }
+                rs_date.close();
+
+                for (Date date = start.getTime(); start.before(end); start.add(Calendar.DATE, 1), date = start.getTime()) {
+
+                    String period="&period=day&date="+sdf.format(date);
+
+                    String apiUrl=getPiwikLogUrl()+APImethod+"&idSite="+siteId+period+format+"&expanded=5&filter_limit=1000&token_auth="+tokenAuth;
+                    String content = getJson(apiUrl,piwikUsername,piwikPassword);
+
+                    //for (int i=1;i<10;i++){
+                    int i=1;
+                    while(true) {
+                        String apiUrlnew=apiUrl+"&filter_offset="+i*1000;
+                        String contentNew = getJson(apiUrlnew,piwikUsername,piwikPassword);
+                        content += contentNew;
+                        i++;
+                        if(contentNew.equals("[]\n")){
+                            break;
+                        }
+                    }
+                    flushString(content, logsPath + "repolog/" + siteId + "_Piwiklog"+sdf.format((date))+".json");
+
+                }
+            }
+            rs.close();
+            conn.close();
+        } catch (Exception e) {
+            log.error(e);
+            throw new Exception("Failed to get repository logs: " + e.toString(), e);
+        }
+    }
+
+    private void flushString(String data, String destination) throws Exception {
+        FSDataOutputStream fin;
+        try {
+            FileSystem fs = FileSystem.get(new Configuration());
+            fin = fs.create(new Path(destination), true);
+            fin.write(data.getBytes());
+            fin.close();
+        } catch (Exception e) {
+            log.error("Failed  to write exported data to a file : ", e);
+            throw new Exception("Failed  to write exported data to a file : " + e.toString(), e);
+        }
+    }
+}
Index: modules/dnet-openaire-usage-stats-export/trunk/pom.xml
===================================================================
--- modules/dnet-openaire-usage-stats-export/trunk/pom.xml	(nonexistent)
+++ modules/dnet-openaire-usage-stats-export/trunk/pom.xml	(revision 47499)
@@ -0,0 +1,45 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+    <parent>
+        <groupId>eu.dnetlib</groupId>
+        <artifactId>dnet-parent</artifactId>
+        <version>1.0.0</version>
+        <relativePath></relativePath>
+    </parent>
+
+    <modelVersion>4.0.0</modelVersion>
+    <groupId>eu.dnetlib</groupId>
+    <artifactId>dnet-openaire-usage-stats</artifactId>
+    <version>1.0.0</version>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.googlecode.json-simple</groupId>
+            <artifactId>json-simple</artifactId>
+            <version>1.1.1</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <version>${hadoop.common.version}</version>
+        </dependency>
+    </dependencies>
+
+	<repositories>
+		<repository>
+			<id>cloudera</id>
+			<name>Cloudera Repository</name>
+			<url>https://repository.cloudera.com/artifactory/cloudera-repos</url>
+			<releases>
+				<enabled>true</enabled>
+			</releases>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+		</repository>
+	</repositories>
+
+</project>
\ No newline at end of file
