diff --git a/dhp-workflows/dhp-usage-stats-update/src/main/java/eu/dnetlib/oa/graph/usagestats/export/SarcStats.java b/dhp-workflows/dhp-usage-stats-update/src/main/java/eu/dnetlib/oa/graph/usagestats/export/SarcStats.java
index 2dc2731c7..aabbc4c4e 100644
--- a/dhp-workflows/dhp-usage-stats-update/src/main/java/eu/dnetlib/oa/graph/usagestats/export/SarcStats.java
+++ b/dhp-workflows/dhp-usage-stats-update/src/main/java/eu/dnetlib/oa/graph/usagestats/export/SarcStats.java
@@ -142,7 +142,7 @@ public class SarcStats {
 		System.out.println("====> Created sarc_sushilogtmp_json_non_array_" + issn.replace("-", "_") + " table");
 
 		System.out.println("====> Creating sarc_sushilogtmp table");
-		String create_sarc_sushilogtmp = "CREATE TABLE " + ConnectDB.getUsageStatsDBSchema()
+		String create_sarc_sushilogtmp = "CREATE TABLE IF NOT EXISTS " + ConnectDB.getUsageStatsDBSchema()
 			+ ".sarc_sushilogtmp(source STRING, repository STRING, " +
 			"rid STRING, date STRING, metric_type STRING, count INT) clustered by (source) into 100 buckets stored as orc "
 			+
@@ -150,26 +150,6 @@ public class SarcStats {
 		stmt.executeUpdate(create_sarc_sushilogtmp);
 		System.out.println("====> Created sarc_sushilogtmp table");
 
-//		"INSERT INTO sushilog (source, repository, rid, date, metric_type, count) VALUES (?,?,?,?,?,?)");
-
-//		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));
-
 		System.out.println("====> Inserting to sarc_sushilogtmp table (sarc_sushilogtmp_json_array)");
 		String insert_sarc_sushilogtmp = "INSERT INTO  " + ConnectDB.getUsageStatsDBSchema() + ".sarc_sushilogtmp " +
 			"SELECT 'SARC-OJS', '" + issn + "' , `ItemIdent`.`Value`, `ItemPerformance`.`Period`.`Begin`, " +
@@ -189,6 +169,10 @@ public class SarcStats {
 		System.out.println("====> Inserted to sarc_sushilogtmp table (sarc_sushilogtmp_json_non_array)");
 
 		ConnectDB.getConnection().close();
+
+		////////////////////////////////////
+		// Add everything to the sushilog table!!!!
+		////////////////////////////////////
 	}
 
 	public void getAndProcessSarc(String sarcsReportPathArray, String sarcsReportPathNonArray) throws Exception {
@@ -204,18 +188,18 @@ public class SarcStats {
 		ConnectDB.getConnection().close();
 
 		List<String[]> issnAndUrls = new ArrayList<String[]>();
-		issnAndUrls.add(new String[] {
-			"https://revistas.rcaap.pt/motricidade/sushiLite/v1_7/", "1646-107X"
-		});
-		issnAndUrls.add(new String[] {
-			"https://revistas.rcaap.pt/antropologicas/sushiLite/v1_7/", "0873-819X"
-		});
-		issnAndUrls.add(new String[] {
-			"https://revistas.rcaap.pt/interaccoes/sushiLite/v1_7/", "1646-2335"
-		});
-		issnAndUrls.add(new String[] {
-			"https://revistas.rcaap.pt/cct/sushiLite/v1_7/", "2182-3030"
-		});
+//		issnAndUrls.add(new String[] {
+//			"https://revistas.rcaap.pt/motricidade/sushiLite/v1_7/", "1646-107X"
+//		});
+//		issnAndUrls.add(new String[] {
+//			"https://revistas.rcaap.pt/antropologicas/sushiLite/v1_7/", "0873-819X"
+//		});
+//		issnAndUrls.add(new String[] {
+//			"https://revistas.rcaap.pt/interaccoes/sushiLite/v1_7/", "1646-2335"
+//		});
+//		issnAndUrls.add(new String[] {
+//			"https://revistas.rcaap.pt/cct/sushiLite/v1_7/", "2182-3030"
+//		});
 		issnAndUrls.add(new String[] {
 			"https://actapediatrica.spp.pt/sushiLite/v1_7/", "0873-9781"
 		});
@@ -315,9 +299,28 @@ public class SarcStats {
 				"INSERT INTO sushilog (source, repository, rid, date, metric_type, count) VALUES (?,?,?,?,?,?)");
 		int batch_size = 0;
 
+		// Creating the needed configuration for the correct storing of data
+		Configuration config = new Configuration();
+		config.addResource(new Path("/etc/hadoop/conf/core-site.xml"));
+		config.addResource(new Path("/etc/hadoop/conf/hdfs-site.xml"));
+		config
+			.set(
+				"fs.hdfs.impl",
+				org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
+		config
+			.set(
+				"fs.file.impl",
+				org.apache.hadoop.fs.LocalFileSystem.class.getName());
+		FileSystem dfs = FileSystem.get(config);
+
 		// Creating the directory to save the files
-		new File(sarcsReportPathArray + "/" + issn).mkdirs();
-		new File(sarcsReportPathNonArray + "/" + issn).mkdirs();
+		dfs.mkdirs(new Path(sarcsReportPathArray + "/" + issn));
+		dfs.mkdirs(new Path(sarcsReportPathNonArray + "/" + issn));
+
+//		System.out.println("====> END  ===== ");
+//
+//		if (true)
+//			System.exit(0);
 
 		while (start.before(end)) {
 			// String reportUrl =
@@ -372,18 +375,18 @@ public class SarcStats {
 			}
 
 			// Creating the file in the filesystem for the ItemIdentifier as array object
-			FileSystem fsArray = FileSystem.get(new Configuration());
+//			FileSystem fsArray = FileSystem.get(new Configuration());
 			String filePathArray = sarcsReportPathArray + "/" + issn + "/" + "SarcsARReport" +
 				simpleDateFormat.format(start.getTime()) + ".json";
 			System.out.println("Storing to file: " + filePathArray);
-			FSDataOutputStream finArray = fsArray.create(new Path(filePathArray), true);
+			FSDataOutputStream finArray = dfs.create(new Path(filePathArray), true);
 
 			// Creating the file in the filesystem for the ItemIdentifier as array object
-			FileSystem fsNonArray = FileSystem.get(new Configuration());
+//			FileSystem fsNonArray = FileSystem.get(new Configuration());
 			String filePathNonArray = sarcsReportPathNonArray + "/" + issn + "/" + "SarcsARReport" +
 				simpleDateFormat.format(start.getTime()) + ".json";
 			System.out.println("Storing to file: " + filePathNonArray);
-			FSDataOutputStream finNonArray = fsNonArray.create(new Path(filePathNonArray), true);
+			FSDataOutputStream finNonArray = dfs.create(new Path(filePathNonArray), true);
 
 			for (Object aJsonArray : jsonArray) {
 
@@ -411,6 +414,9 @@ public class SarcStats {
 			if (fileNonArray.length() == 0)
 				fileNonArray.delete();
 
+//			fsArray.close();
+//			fsNonArray.close();
+
 //////////////////
 //			JSONObject jsonObjectRow = (JSONObject) aJsonArray;
 //			JSONArray itemIdentifier = new JSONArray();
@@ -465,6 +471,8 @@ public class SarcStats {
 			// break;
 		}
 
+		dfs.close();
+
 //		preparedStatement.executeBatch();
 //		ConnectDB.getConnection().commit();
 		ConnectDB.getConnection().close();