D-Net

public class PiwikStatsDB {

    private String logRepoPath;

    private String logPortalPath;

    private String CounterRobotsURL;

    private ArrayList robotsList;

    public PiwikStatsDB(String logRepoPath, String logPortalPath) throws Exception {

        this.logRepoPath = logRepoPath;

        this.logPortalPath = logPortalPath;

    }

    public void foo() {

        System.out.println(s.parallel().count());

    }

            this.robotsList = counterRobots.getRobotsPatterns();

            processRepositoryLog();

            removeDoubleClicks();

            processPortalLog();

            log.info("portal process done");

            portalStats();

            log.info("portal usagestats done");

            updateProdTables();

            log.info("updateProdTables done");

//            throw new Exception("Failed to create usage usagestats: " + e.toString(), e);

//        }

//    }

    public void processRepositoryLog() throws Exception {

        Statement stmt = ConnectDB.getConnection().createStatement();

        ConnectDB.getConnection().setAutoCommit(false);

        sql = "CREATE TABLE IF NOT EXISTS pageviews_stats (like pageviews_stats_tmp including all)";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.getConnection().commit();

        ConnectDB.getConnection().close();

//        ConnectDB.getConnection().commit();

//        ConnectDB.getConnection().close();

//    }

        Statement stmt = ConnectDB.getConnection().createStatement();

        ConnectDB.getConnection().setAutoCommit(false);

        String sql = "CREATE OR REPLACE VIEW result_downloads_monthly_tmp 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 piwiklogtmp 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, 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;";

        sql = "CREATE TABLE IF NOT EXISTS downloads_stats_tmp AS 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 FROM result_downloads_monthly_tmp p, public.datasource d, public.result_oids ro where 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 = "CREATE TABLE IF NOT EXISTS downloads_stats (like downloads_stats_tmp including all)";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.getConnection().commit();

        ConnectDB.getConnection().close();

        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 IF NOT EXISTS full_dates_full_date ON full_dates USING btree(full_date);";

        stmt.executeUpdate(sql);

//        sql = "SELECT coalesce(ds.source, vs.source) as source, coalesce(ds.repository_id, vs.repository_id) as repository_id, coalesce(ds.result_id, vs.result_id) as result_id, coalesce(ds.date, vs.date) as date, coalesce(ds.count, 0) as downloads, coalesce(vs.count, 0) as views, coalesce(ds.openaire, 0) as openaire_downloads, coalesce(vs.openaire, 0) as openaire_views INTO usage_stats FROM downloads_stats AS ds FULL OUTER JOIN views_stats AS vs ON ds.source=vs.source AND ds.repository_id=vs.repository_id AND ds.result_id=vs.result_id AND ds.date=vs.date;";

        sql = "CREATE TABLE IF NOT EXISTS usage_stats AS SELECT coalesce(ds.source, vs.source) as source, coalesce(ds.repository_id, vs.repository_id) as repository_id, coalesce(ds.result_id, vs.result_id) as result_id, coalesce(ds.date, vs.date) as date, coalesce(ds.count, 0) as downloads, coalesce(vs.count, 0) as views, coalesce(ds.openaire, 0) as openaire_downloads, coalesce(vs.openaire, 0) as openaire_views FROM downloads_stats AS ds FULL OUTER JOIN views_stats AS vs ON ds.source=vs.source AND ds.repository_id=vs.repository_id AND ds.result_id=vs.result_id AND ds.date=vs.date;";

        stmt.executeUpdate(sql);

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX IF NOT EXISTS usage_stats_source ON usage_stats USING btree(source);";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX IF NOT EXISTS usage_stats_date ON usage_stats USING btree(date);";

        stmt.executeUpdate(sql);

        stmt.executeUpdate(sql);

        stmt.executeUpdate(sql);

        stmt.executeUpdate(sql);

        sql = "DROP TABLE IF EXISTS downloads_stats_tmp;";

        stmt.executeUpdate(sql);

        sql = "DROP TABLE IF EXISTS process_portal_log_tmp;";

        stmt.executeUpdate(sql);

        sql = "DROP TABLE IF EXISTS sushilogtmp;";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.getConnection().commit();

        ConnectDB.getConnection().close();

        stmt.close();

        ConnectDB.getConnection().close();

    }

    public void portalStats() throws SQLException {

        Connection con = ConnectDB.getConnection();

        Statement stmt = con.createStatement();

        stmt = con.createStatement();

        sql = "INSERT INTO piwiklogtmp SELECT DISTINCT source, id_visit, country, action, url, roid.orid, \'project\', timestamp, referrer_name, agent FROM process_portal_log_tmp, public.project_oids roid WHERE entity_id IS NOT null AND entity_id=roid.orid AND roid.orid IS NOT null;";

        stmt.executeUpdate(sql);

//        con.commit();

        stmt.close();

        ConnectDB.getConnection().commit();

        Statement stmt = ConnectDB.getConnection().createStatement();

        ConnectDB.getConnection().setAutoCommit(false);

        sql = "insert into views_stats select * from views_stats_tmp;";

        stmt.executeUpdate(sql);

        sql = "insert into downloads_stats select * from downloads_stats_tmp;";

        stmt.executeUpdate(sql);

        sql = "insert into pageviews_stats select * from pageviews_stats_tmp;";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.getConnection().commit();

        ConnectDB.getConnection().close();

            hdfs.close();

        } catch (Exception e) {

        }

        return fileNames;