Feature/hdfs dev

CMakeLists.txt

@@ -79,6 +79,9 @@ set(HEADERS globals.h
         src/streamingdb/StreamingSQLiteDBInterface.h
         src/frontend/core/executor/impl/PageRankExecutor.h
         src/util/dbinterface/DBInterface.h
+        src/util/hdfs/HDFSConnector.h
+        src/util/hdfs/HDFSStreamHandler.h
+        src/partitioner/stream/HashPartitioner.h
         src/query/processor/cypher/util/Const.h
 )
 
@@ -148,6 +151,9 @@ set(SOURCES src/backend/JasmineGraphBackend.cpp
         src/streamingdb/StreamingSQLiteDBInterface.cpp
         src/frontend/core/executor/impl/PageRankExecutor.cpp
         src/util/dbinterface/DBInterface.cpp
+        src/util/hdfs/HDFSConnector.cpp
+        src/util/hdfs/HDFSStreamHandler.cpp
+        src/partitioner/stream/HashPartitioner.cpp
         src/query/processor/cypher/semanticanalyzer/SemanticAnalyzer.cpp
         src/query/processor/cypher/util/Const.cpp
 )
@@ -179,9 +185,14 @@ target_link_libraries(JasmineGraphLib PRIVATE /usr/lib/x86_64-linux-gnu/libxerce
 target_link_libraries(JasmineGraphLib PRIVATE /usr/lib/x86_64-linux-gnu/libflatbuffers.a)
 target_link_libraries(JasmineGraphLib PRIVATE /usr/lib/x86_64-linux-gnu/libjsoncpp.so)
 target_link_libraries(JasmineGraphLib PRIVATE /usr/local/lib/libcppkafka.so)
+target_link_libraries(JasmineGraphLib PRIVATE /usr/local/libhdfs3/lib/libhdfs3.so)
 target_link_libraries(JasmineGraph JasmineGraphLib)
 target_link_libraries(JasmineGraph curl)
 
+include_directories(/usr/local/libhdfs3/include/hdfs/)
+include_directories(/usr/include/oneapi/)
+include_directories(/usr/local/parmetis/include/)
+include_directories(/usr/lib/x86_64-linux-gnu/openmpi/include/)
 include_directories(/usr/local/include/antlr4-runtime)
 link_directories(/usr/local/lib)
 include_directories(/usr/local/include/yaml-cpp)

Dockerfile

@@ -1,4 +1,4 @@
-FROM miyurud/jasminegraph-prerequisites:20241230T132919
+FROM miyurud/jasminegraph-prerequisites:20241231T070657
 
 RUN apt-get update && apt-get install -y libcurl4-openssl-dev sysstat nmon
 RUN rm -r /usr/lib/python3.8/distutils

conf/hdfs_config/consumer.properties

@@ -0,0 +1,2 @@
+hdfs.host=127.0.0.1
+hdfs.port=9000

conf/jasminegraph-server.properties

@@ -24,6 +24,8 @@ org.jasminegraph.server.nworkers=2
 #org.jasminegraph.server.npartitions is the number of partitions into which the graph should be partitioned
 org.jasminegraph.server.npartitions=2
 org.jasminegraph.server.streaming.kafka.host=127.0.0.1:9092
+org.jasminegraph.server.streaming.hdfs.host=hdfs://10.8.100.246
+org.jasminegraph.server.streaming.hdfs.port=9000
 org.jasminegraph.worker.path=/var/tmp/jasminegraph/
 #Path to keep jasminegraph artifacts in order to copy them to remote locations. If this is not set then the artifacts in the
 #JASMINEGRAPH_HOME location will be copied instead.
@@ -32,6 +34,7 @@ org.jasminegraph.artifact.path=
 #The following folder is the location where workers keep their data.
 #This is the location where the actual data storage takes place in JasmineGraph.
 org.jasminegraph.server.instance.datafolder=/var/tmp/jasminegraph-localstore
+org.jasminegraph.server.instance.tempdatafolder=/var/tmp/tempfiles
 #The folder path for keeping central stores for triangle count aggregation
 org.jasminegraph.server.instance.aggregatefolder=/var/tmp/jasminegraph-aggregate
 org.jasminegraph.server.instance.trainedmodelfolder=/var/tmp/jasminegraph-localstore/jasminegraph-local_trained_model_store

src/frontend/JasmineGraphFrontEnd.cpp

@@ -45,14 +45,15 @@
 #include "JasmineGraphFrontEndProtocol.h"
 #include "core/CoreConstants.h"
 #include "core/scheduler/JobScheduler.h"
+#include "../util/hdfs/HDFSConnector.h"
+#include "../util/hdfs/HDFSStreamHandler.h"
 #include "antlr4-runtime.h"
 #include "/home/ubuntu/software/antlr/CypherLexer.h"
 #include "/home/ubuntu/software/antlr/CypherParser.h"
 #include "../query/processor/cypher/astbuilder/ASTBuilder.h"
 #include "../query/processor/cypher/astbuilder/ASTNode.h"
 #include "../query/processor/cypher/semanticanalyzer/SemanticAnalyzer.h"
 
-
 #define MAX_PENDING_CONNECTIONS 10
 #define DATA_BUFFER_SIZE (FRONTEND_DATA_LENGTH + 1)
 
@@ -81,6 +82,9 @@
                                      KafkaConnector *&kstream, thread &input_stream_handler_thread,
                                      vector<DataPublisher *> &workerClients, int numberOfPartitions,
                                      SQLiteDBInterface *sqlite, bool *loop_exit_p);
+static void addStreamHDFSCommand(std::string masterIP, int connFd, std::string &hdfsServerIp,
+                                 std::thread &inputStreamHandlerThread, int numberOfPartitions,
+                                 SQLiteDBInterface *sqlite, bool *loop_exit_p);
 static void stop_stream_kafka_command(int connFd, KafkaConnector *kstream, bool *loop_exit_p);
 static void process_dataset_command(int connFd, bool *loop_exit_p);
 static void triangles_command(std::string masterIP, int connFd, SQLiteDBInterface *sqlite,
@@ -109,8 +113,9 @@
     for (int i = 0; i < workerList.size(); i++) {
         Utils::worker currentWorker = workerList.at(i);
         string workerHost = currentWorker.hostname;
+        int workerDataPort = std::stoi(currentWorker.dataPort);
         int workerPort = atoi(string(currentWorker.port).c_str());
-        DataPublisher *workerClient = new DataPublisher(workerPort, workerHost);
+        DataPublisher *workerClient = new DataPublisher(workerPort, workerHost, workerDataPort);
         workerClients.push_back(workerClient);
     }
     return workerClients;
@@ -144,6 +149,10 @@
     KafkaConnector *kstream;
     Partitioner graphPartitioner(numberOfPartitions, 1, spt::Algorithms::HASH, sqlite);
 
+    // Initiate HDFS parameters
+    std::string hdfsServerIp;
+    hdfsFS fileSystem;
+
     vector<DataPublisher *> workerClients;
     bool workerClientsInitialized = false;
 
@@ -198,6 +207,9 @@
             }
             add_stream_kafka_command(connFd, kafka_server_IP, configs, kstream, input_stream_handler, workerClients,
                                      numberOfPartitions, sqlite, &loop_exit);
+        } else if (line.compare(ADD_STREAM_HDFS) == 0) {
+            addStreamHDFSCommand(masterIP, connFd, hdfsServerIp, input_stream_handler, numberOfPartitions,
+                                    sqlite, &loop_exit);
         } else if (line.compare(STOP_STREAM_KAFKA) == 0) {
             stop_stream_kafka_command(connFd, kstream, &loop_exit);
         } else if (line.compare(RMGR) == 0) {
@@ -1304,8 +1316,186 @@
     input_stream_handler_thread = thread(&StreamHandler::listen_to_kafka_topic, stream_handler);
 }
 
+void addStreamHDFSCommand(std::string masterIP, int connFd, std::string &hdfsServerIp,
+                             std::thread &inputStreamHandlerThread, int numberOfPartitions,
+                             SQLiteDBInterface *sqlite, bool *loop_exit_p) {
+    std::string hdfsPort;
+    std::string msg1 = "Do you want to use the default HDFS server(y/n)?";
+    int resultWr = write(connFd, msg1.c_str(), msg1.length());
+    if (resultWr < 0) {
+        frontend_logger.error("Error writing to socket");
+        *loop_exit_p = true;
+        return;
+    }
+    resultWr = write(connFd, "\r\n", 2);
+    if (resultWr < 0) {
+        frontend_logger.error("Error writing to socket");
+        *loop_exit_p = true;
+        return;
+    }
+
+    char userRes[FRONTEND_DATA_LENGTH + 1];
+    bzero(userRes, FRONTEND_DATA_LENGTH + 1);
+    read(connFd, userRes, FRONTEND_DATA_LENGTH);
+    std::string userResS(userRes);
+    userResS = Utils::trim_copy(userResS);
+    for (char &c : userResS) {
+        c = tolower(c);
+    }
+
+    if (userResS == "y") {
+        hdfsServerIp = Utils::getJasmineGraphProperty("org.jasminegraph.server.streaming.hdfs.host");
+        hdfsPort = Utils::getJasmineGraphProperty("org.jasminegraph.server.streaming.hdfs.port");
+    } else {
+        std::string message = "Send the file path to the HDFS configuration file.";
+        resultWr = write(connFd, message.c_str(), message.length());
+        if (resultWr < 0) {
+            frontend_logger.error("Error writing to socket");
+            *loop_exit_p = true;
+            return;
+        }
+        resultWr = write(connFd, "\r\n", 2);
+        if (resultWr < 0) {
+            frontend_logger.error("Error writing to socket");
+            *loop_exit_p = true;
+            return;
+        }
+
+        char filePath[FRONTEND_DATA_LENGTH + 1];
+        bzero(filePath, FRONTEND_DATA_LENGTH + 1);
+        read(connFd, filePath, FRONTEND_DATA_LENGTH);
+        std::string filePathS(filePath);
+        filePathS = Utils::trim_copy(filePathS);
+
+        frontend_logger.info("Reading HDFS configuration file: " + filePathS);
+
+        std::vector<std::string> vec = Utils::getFileContent(filePathS);
+        for (const auto &item : vec) {
+            if (item.length() > 0 && !(item.rfind("#", 0) == 0)) {
+                std::vector<std::string> vec2 = Utils::split(item, '=');
+                if (vec2.size() == 2) {
+                    if (vec2.at(0).compare("hdfs.host") == 0) {
+                        hdfsServerIp = vec2.at(1);
+                    } else if (vec2.at(0).compare("hdfs.port") == 0) {
+                        hdfsPort = vec2.at(1);
+                    }
+                } else {
+                    frontend_logger.error("Invalid line in configuration file: " + item);
+                }
+            }
+        }
+    }
+
+    if (hdfsServerIp.empty()) {
+        frontend_logger.error("HDFS server IP is not set or empty.");
+    }
+    if (hdfsPort.empty()) {
+        frontend_logger.error("HDFS server port is not set or empty.");
+    }
+
+    std::string message = "HDFS file path: ";
+    resultWr = write(connFd, message.c_str(), message.length());
+    if (resultWr < 0) {
+        frontend_logger.error("Error writing to socket");
+        *loop_exit_p = true;
+        return;
+    }
+    resultWr = write(connFd, "\r\n", 2);
+    if (resultWr < 0) {
+        frontend_logger.error("Error writing to socket");
+        *loop_exit_p = true;
+        return;
+    }
+
+    char hdfsFilePath[FRONTEND_DATA_LENGTH + 1];
+    bzero(hdfsFilePath, FRONTEND_DATA_LENGTH + 1);
+    read(connFd, hdfsFilePath, FRONTEND_DATA_LENGTH);
+    std::string hdfsFilePathS(hdfsFilePath);
+    hdfsFilePathS = Utils::trim_copy(hdfsFilePathS);
+
+    HDFSConnector *hdfsConnector = new HDFSConnector(hdfsServerIp, hdfsPort);
+
+    if (!hdfsConnector->isPathValid(hdfsFilePathS)) {
+        frontend_logger.error("Invalid HDFS file path: " + hdfsFilePathS);
+        std::string error_message = "The provided HDFS path is invalid.";
+        write(connFd, error_message.c_str(), error_message.length());
+        write(connFd, "\r\n", 2);
+        delete hdfsConnector;
+        *loop_exit_p = true;
+        return;
+    }
+
+    /*get directionality*/
+    std::string isDirectedGraph = "Is this a directed graph(y/n)?";
+    resultWr = write(connFd, isDirectedGraph.c_str(), isDirectedGraph.length());
+    if (resultWr < 0) {
+        frontend_logger.error("Error writing to socket");
+        *loop_exit_p = true;
+        return;
+    }
+    resultWr = write(connFd, "\r\n", 2);
+    if (resultWr < 0) {
+        frontend_logger.error("Error writing to socket");
+        *loop_exit_p = true;
+        return;
+    }
+
+    char isDirectedRes[FRONTEND_DATA_LENGTH + 1];
+    bzero(isDirectedRes, FRONTEND_DATA_LENGTH + 1);
+    read(connFd, isDirectedRes, FRONTEND_DATA_LENGTH);
+    std::string isDirectedS(isDirectedRes);
+    isDirectedS = Utils::trim_copy(isDirectedS);
+
+    bool directed;
+    if (isDirectedS == "y") {
+        directed = true;
+    } else {
+        directed = false;
+    }
+
+    std::string path = "hdfs:" + hdfsFilePathS;
+
+    std::time_t time = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
+    std::string uploadStartTime = ctime(&time);
+    std::string sqlStatement =
+            "INSERT INTO graph (name, upload_path, upload_start_time, upload_end_time, graph_status_idgraph_status, "
+            "vertexcount, centralpartitioncount, edgecount, is_directed) VALUES(\"" +
+            hdfsFilePathS + "\", \"" + path + "\", \"" + uploadStartTime + "\", \"\", \"" +
+            std::to_string(Conts::GRAPH_STATUS::NONOPERATIONAL) + "\", \"\", \"\", \"\", \"" +
+            (directed ? "TRUE" : "FALSE") + "\")";
+
+    int newGraphID = sqlite->runInsert(sqlStatement);
+    frontend_logger.info("Created graph ID: " + std::to_string(newGraphID));
+    HDFSStreamHandler *streamHandler = new HDFSStreamHandler(hdfsConnector->getFileSystem(), hdfsFilePathS,
+                                                              numberOfPartitions, newGraphID, sqlite, masterIP);
+    frontend_logger.info("Started listening to " + hdfsFilePathS);
+    inputStreamHandlerThread = std::thread(&HDFSStreamHandler::startStreamingFromBufferToPartitions, streamHandler);
+    inputStreamHandlerThread.join();
+
+    std::string uploadEndTime = ctime(&time);
+    std::string sqlStatementUpdateEndTime =
+            "UPDATE graph "
+            "SET upload_end_time = \"" + uploadEndTime + "\" "
+                                                         "WHERE idgraph = " + std::to_string(newGraphID);
+    sqlite->runInsert(sqlStatementUpdateEndTime);
+
+
+    int conResultWr = write(connFd, DONE.c_str(), DONE.length());
+    if (conResultWr < 0) {
+        frontend_logger.error("Error writing to socket");
+        *loop_exit_p = true;
+        return;
+    }
+    resultWr = write(connFd, "\r\n", 2);
+    if (resultWr < 0) {
+        frontend_logger.error("Error writing to socket");
+        *loop_exit_p = true;
+        return;
+    }
+}
+
 static void stop_stream_kafka_command(int connFd, KafkaConnector *kstream, bool *loop_exit_p) {
-    frontend_logger.info("Start serving `" + STOP_STREAM_KAFKA + "` command");
+    frontend_logger.info("Started serving `" + STOP_STREAM_KAFKA + "` command");
     //          Unsubscribe the kafka consumer.
     kstream->Unsubscribe();
     string message = "Successfully stop `" + stream_topic_name + "` input kafka stream";

src/frontend/JasmineGraphFrontEndProtocol.cpp

@@ -24,6 +24,7 @@ const string SHTDN = "shdn";
 const string SEND = "send";
 const string ERROR = "error";
 const string ADD_STREAM_KAFKA = "adstrmk";
+const string ADD_STREAM_HDFS = "adhdfs";
 const string ADD_STREAM_KAFKA_CSV = "adstrmkcsv";
 const string STOP_STREAM_KAFKA = "stopstrm";
 const string STOP_STRIAN = "stopstrian";

src/frontend/JasmineGraphFrontEndProtocol.h

@@ -62,6 +62,7 @@ extern const string OUTPUT_FILE_NAME;
 extern const string OUTPUT_FILE_PATH;
 extern const string ADD_STREAM;
 extern const string ADD_STREAM_KAFKA;
+extern const string ADD_STREAM_HDFS;
 extern const string STRM_ACK;
 extern const string ADD_STREAM_KAFKA;
 extern const string STREAM_TOPIC_NAME;

src/localstore/incremental/JasmineGraphIncrementalLocalStore.cpp

@@ -106,3 +106,17 @@
         // TODO tmkasun: handle JSON errors
     }
 }
+
+void JasmineGraphIncrementalLocalStore::addLocalEdge(const std::pair<std::string, std::string> &edge) {
+    RelationBlock* newRelation;
+    newRelation = this->nm->addLocalEdge({edge.first, edge.second});
+    // [ToDo]:implement add edge properties
+    incremental_localstore_logger.debug("Local edge ("+edge.first+"-> "+edge.second+" ) added successfully");
+}
+
+void JasmineGraphIncrementalLocalStore::addCentralEdge(const std::pair<std::string, std::string> &edge) {
+    RelationBlock* newRelation;
+    newRelation = this->nm->addCentralEdge({edge.first, edge.second});
+    // [ToDo]:implement add edge properties
+    incremental_localstore_logger.debug("Central edge ("+edge.first+"-> "+edge.second+" ) added successfully");
+}

src/localstore/incremental/JasmineGraphIncrementalLocalStore.h

@@ -27,6 +27,8 @@ class JasmineGraphIncrementalLocalStore {
     static std::pair<std::string, unsigned int> getIDs(std::string edgeString);
     JasmineGraphIncrementalLocalStore(unsigned int graphID = 0,
                                       unsigned int partitionID = 0, std::string openMode = "trunk");
+    void addLocalEdge(const std::pair<std::string, std::string> &edge);
+    void addCentralEdge(const std::pair<std::string, std::string> &edge);
 };
 
 #endif

src/nativestore/DataPublisher.cpp

@@ -21,9 +21,10 @@
 
 Logger data_publisher_logger;
 
-DataPublisher::DataPublisher(int worker_port, std::string worker_address) {
+DataPublisher::DataPublisher(int worker_port, std::string worker_address, int worker_data_port) {
     this->worker_port = worker_port;
     this->worker_address = worker_address;
+    this->data_port = worker_data_port;
     struct hostent *server;
 
     server = gethostbyname(worker_address.c_str());
@@ -42,6 +43,7 @@
     if (Utils::connect_wrapper(sock, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) < 0) {
         data_publisher_logger.error("Connection Failed!");
     }
+    data_publisher_logger.info("socket created");
 }
 
 DataPublisher::~DataPublisher() {

src/nativestore/DataPublisher.h

@@ -31,9 +31,10 @@ class DataPublisher {
     struct sockaddr_in serv_addr;
     std::string worker_address, message;
     char buffer[1024] = {0};
+    int data_port;
 
  public:
-    DataPublisher(int, std::string);
+    DataPublisher(int, std::string, int);
     void publish(std::string);
     void publish_relation(std::string);
     void publish_edge(std::string);