D-Net

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 IrusStats {

     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 IrusStats(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, rid TEXT, date TEXT, metric_type TEXT, count INT, PRIMARY KEY(source, repository, rid, 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.rid, sushilog.date " +

                "FROM sushilog " +

                "WHERE sushilog.source = new.source AND sushilog.repository = new.repository AND sushilog.rid = new.rid 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 irusStats() 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.rid=ro.orid AND metric_type='ft_total' AND s.source='IRUS-UK';";

        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, rid, 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);

        }

    }

    */

}

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 27/6/2017.

 */

public class SarcStats {

    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 SarcStats(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, rid TEXT, date TEXT, metric_type TEXT, count INT, PRIMARY KEY(source, repository, rid, 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.rid, sushilog.date " +

                    "FROM sushilog " +

                    "WHERE sushilog.source = new.source AND sushilog.repository = new.repository AND sushilog.rid = new.rid 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 processSarc() throws Exception {

        processARReport("https://revistas.rcaap.pt/motricidade/sushiLite/v1_7/", "1646-107X");

        processARReport("http://revistas.rcaap.pt/antropologicas/sushiLite/v1_7/", "0873-819X");

        processARReport("https://revistas.rcaap.pt/interaccoes/sushiLite/v1_7/", "1646-2335");

        processARReport("http://revistas.rcaap.pt/cct/sushiLite/v1_7/", "2182-3030");

        processARReport("http://actapediatrica.spp.pt/sushiLite/v1_7/", "0873-9781");

        //processARReport("http://revistas.rcaap.pt/boletimspm/sushiLite/v1_7/");

        processARReport("https://revistas.rcaap.pt/sociologiapp/sushiLite/v1_7/", "0873-6529");

        processARReport("http://revistas.rcaap.pt/finisterra/sushiLite/v1_7/", "0430-5027");

        processARReport("http://revistas.rcaap.pt/sisyphus/sushiLite/v1_7/", "2182-8474");

        processARReport("https://revistas.rcaap.pt/anestesiologia/sushiLite/v1_7/", "0871-6099");

        processARReport("https://revistas.rcaap.pt/rpe/sushiLite/v1_7/", "0871-9187");

        processARReport("https://revistas.rcaap.pt/psilogos/sushiLite/v1_7/", "1646-091X");

        processARReport("https://revistas.rcaap.pt/juridica/sushiLite/v1_7/", "2183-5799");

        processARReport("http://revistas.rcaap.pt/ecr/sushiLite/v1_7/", "1647-2098");

        processARReport("https://revistas.rcaap.pt/nascercrescer/sushiLite/v1_7/", "0872-0754");

        processARReport("http://revistas.rcaap.pt/cea/sushiLite/v1_7/", "1645-3794");

        processARReport("http://revistas.rcaap.pt/proelium/sushiLite/v1_7/", "1645-8826");

        processARReport("http://revistas.rcaap.pt/millenium/sushiLite/v1_7/", "0873-3015");

        //processARReport("http://www.bad.pt/publicacoes/index.php/cadernos/sushiLite/v1_7/");

    }

    public void sarcStats() throws Exception {

        if (conn.isClosed())

            connectDB();

        stmt = conn.createStatement();

        conn.setAutoCommit(false);

        //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, '0' INTO downloads_stats FROM sushilog s, datasource_oids d, result_oids ro WHERE s.repository=d.orid AND s.rid=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, datasource_results dr, result_pids ro WHERE d.orid LIKE '%' || s.repository || '%' AND dr.id=d.id AND dr.result=ro.id AND s.rid=ro.pid AND ro.type='doi' AND metric_type='ft_total' AND s.source='SARC-OJS';";

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        conn.close();

    }

    public void processARReport(String url, String issn) throws Exception {

        log.info("Processing SARC! issn: " + issn + " with url: " + url);

        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, issn);

        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, rid, date, metric_type, count) VALUES (?,?,?,?,?,?)");

        int batch_size = 0;

        while(start.before(end)){

            //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=";

            String reportUrl = url + "GetReport/?Report=AR1&Format=json&BeginDate=" + simpleDateFormat.format(start.getTime()) + "&EndDate=" + simpleDateFormat.format(start.getTime());

            //System.out.println(reportUrl);

            start.add(Calendar.MONTH, 1);

            String text = getJson(reportUrl, "", "");

            if(text == null){

                continue;

            }

            /*

            PrintWriter wr = new PrintWriter(new FileWriter("logs/" + simpleDateFormat.format(start.getTime()) + ".json"));

            wr.print(text);

            wr.close();

            */

            JSONParser parser = new JSONParser();

            JSONObject jsonObject = (JSONObject) parser.parse(text);

            jsonObject = (JSONObject) jsonObject.get("sc:ReportResponse");

            jsonObject = (JSONObject) jsonObject.get("sc:Report");

            if(jsonObject == null){

                continue;

            }

            jsonObject = (JSONObject) jsonObject.get("c:Report");

            jsonObject = (JSONObject) jsonObject.get("c:Customer");

            Object obj = jsonObject.get("c:ReportItems");

            JSONArray jsonArray = new JSONArray();

            if(obj instanceof JSONObject){

                jsonArray.add(obj);

            }

            else{

                jsonArray = (JSONArray) obj;

                //jsonArray = (JSONArray) jsonObject.get("c:ReportItems");

            }

            if(jsonArray == null){

                continue;

            }

            String rid = "";

            for (Object aJsonArray : jsonArray) {

                JSONObject jsonObjectRow = (JSONObject) aJsonArray;

                JSONArray itemIdentifier = new JSONArray();

                obj = jsonObjectRow.get("c:ItemIdentifier");

                if(obj instanceof JSONObject){

                    itemIdentifier.add(obj);

                }

                else{

                    //JSONArray itemIdentifier = (JSONArray) jsonObjectRow.get("c:ItemIdentifier");

                    itemIdentifier = (JSONArray) obj;

                }

                for (Object identifier : itemIdentifier) {

                    JSONObject doi = (JSONObject) identifier;

                    if(doi.get("c:Type").toString().equals("DOI")){

                        rid = doi.get("c:Value").toString();

                        //System.out.println("DOI: " + rid);

                        break;

                    }

                }

                if(rid.isEmpty()){

                    continue;

                }

                JSONObject itemPerformance = (JSONObject) jsonObjectRow.get("c:ItemPerformance");

                //for (Object perf : itemPerformance) {

                JSONObject performance = (JSONObject) itemPerformance;

                JSONObject periodObj = (JSONObject) performance.get("c:Period");

                String period = periodObj.get("c:Begin").toString();

                JSONObject instanceObj = (JSONObject) performance.get("c:Instance");

                String type = instanceObj.get("c:MetricType").toString();

                String count = instanceObj.get("c:Count").toString();

                //System.out.println(rid + " : " + period + " : " + count);

                preparedStatement.setString(1, "SARC-OJS");

                preparedStatement.setString(2, issn);

                //preparedStatement.setString(2, url);

                preparedStatement.setString(3, rid);

                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);

        }

    }

    */

}