Importare dati in FalkorDB da MariaDB con Java

FalkorDB è un graph database compatibile con Neo4j.

In questo articolo vediamo come importare i dati prendendoli da MariaDB usando Java.

Io ho installato FalkorDB in Docker:

docker run -p 6379:6379 -p 3000:3000 -v $(pwd)/falkordb_data:/var/lib/falkordb/data --name falkor-test -d falkordb/falkordb:latest

Se usate Maven aggiungete queste due librerie:

<dependencies>
    <dependency>
        <groupId>org.mariadb.jdbc</groupId>
        <artifactId>mariadb-java-client</artifactId>
        <version>3.3.3</version>
    </dependency>

    <dependency>
        <groupId>com.falkordb</groupId>
        <artifactId>jfalkordb</artifactId>
        <version>0.7.0</version>
        <scope>compile</scope>
    </dependency>
</dependencies>

Poi creiamo una classe wrapper:

package org.example;

public class Movimento {

    private String movTipo;
    private double movValore;
    private int anno;
    private String causaleNome;
    private String contoNome;
    private String contoIban;

    public Movimento(String movTipo, double movValore, int anno, String causaleNome, String contoNome, String contoIban) {
        this.movTipo = movTipo;
        this.movValore = movValore;
        this.anno = anno;
        this.causaleNome = causaleNome;
        this.contoNome = contoNome;
        this.contoIban = contoIban;
    }

    public String getMovTipo() {
        return movTipo;
    }

    public double getMovValore() {
        return movValore;
    }

    public int getAnno() {
        return anno;
    }

    public String getCausaleNome() {
        return causaleNome;
    }

    public String getContoNome() {
        return contoNome;
    }

    public String getContoIban() {
        return contoIban;
    }
}

Volendo potreste usare anche un tipo record.

Qui sotto un pò di codice:

package org.example;

import com.falkordb.Driver;
import com.falkordb.FalkorDB;
import com.falkordb.Graph;
import com.falkordb.Record;
import com.falkordb.ResultSet;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class Main {

    private static final String MARIADB_URL = "jdbc:mariadb://localhost:3306/Sql592686_5";
    private static final String MARIADB_USER = "root";
    private static final String MARIADB_PASS = "9211";
    private static final String FALKOR_HOST = "localhost";
    private static final int FALKOR_PORT = 6379;
    private static final String GRAPH_NAME = "movimenti_graph";

    public static void main(String[] args) throws IOException {
        List<Movimento> movimenti = fetchFromMariaDB();
        loadIntoFalkor(movimenti);
        queryFalkor();
    }

    private static List<Movimento> fetchFromMariaDB() {
        System.out.println("Connessione a MariaDB...");

        String MARIADB_QUERY = "SELECT mov_tipo," +
                "       mov_valore," +
                "       YEAR(mov_data) AS anno," +
                "       causale_nome," +
                "       conto_nome," +
                "       conto_iban" +
                " FROM movimenti" +
                "   INNER JOIN causali ON causale_id = mov_causale_fk" +
                "   INNER JOIN conti   ON conto_id   = mov_conto_fk";

        List<Movimento> list = new ArrayList<>();

        try (Connection conn = DriverManager.getConnection(MARIADB_URL, MARIADB_USER, MARIADB_PASS);
             Statement stmt = conn.createStatement();
             java.sql.ResultSet rs = stmt.executeQuery(MARIADB_QUERY)) {

            while (rs.next()) {
                Movimento m = new Movimento(
                        rs.getString("mov_tipo"),
                        rs.getDouble("mov_valore"),
                        rs.getInt("anno"),
                        rs.getString("causale_nome"),
                        rs.getString("conto_nome"),
                        rs.getString("conto_iban")
                );
                list.add(m);
            }
            System.out.println("Recuperati " + list.size() + " record da MariaDB");

        } catch (SQLException e) {
            System.err.println("Errore MariaDB: " + e.getMessage());
            throw new RuntimeException(e);
        }
        return list;
    }

    private static void loadIntoFalkor(List<Movimento> movimenti) throws IOException {
        System.out.println("Connessione a FalkorDB...");

        try (Driver driver = FalkorDB.driver(FALKOR_HOST, FALKOR_PORT)) {
            try {
                driver.graph(GRAPH_NAME).deleteGraph();
                System.out.println("Grafo precedente eliminato");
            } catch (Exception ignored) {
            }

            Graph graph = driver.graph(GRAPH_NAME);
            System.out.println("Caricamento nodi e relazioni...");

            for (Movimento m : movimenti) {
                String causale = escape(m.getCausaleNome());
                String conto = escape(m.getContoNome());
                String iban = escape(m.getContoIban() != null ? m.getContoIban() : "");
                String tipo = escape(m.getMovTipo() != null ? m.getMovTipo() : "");

                graph.query(String.format("MERGE (c:Causale {nome: '%s'})", causale));

                graph.query(String.format("MERGE (ct:Conto {nome: '%s', iban: '%s'})", conto, iban));

                String cypher = String.format(java.util.Locale.US,
                        "MATCH (c:Causale {nome: '%s'})" +
                                " MATCH (ct:Conto {nome: '%s'})" +
                                " CREATE (m:Movimento {tipo: '%s', valore: %f, anno: %d})" +
                                " CREATE (m)-[:HA_CAUSALE]->(c)" +
                                " CREATE (m)-[:SU_CONTO]->(ct)",
                        causale, conto, tipo, m.getMovValore(), m.getAnno());

                graph.query(cypher);
            }

            System.out.println("Caricati " + movimenti.size() + " movimenti nel grafo '" + GRAPH_NAME + "'");
        }
    }

    private static void queryFalkor() throws IOException {
        System.out.println("Query su FalkorDB:");

        try (Driver driver = FalkorDB.driver(FALKOR_HOST, FALKOR_PORT)) {
            Graph graph = driver.graph(GRAPH_NAME);

            System.out.println("-- Totale movimenti per causale --");

            ResultSet rs1 = graph.query(
                    "MATCH (m:Movimento)-[:HA_CAUSALE]->(c:Causale)" +
                            " RETURN c.nome AS causale, COUNT(m) AS num_movimenti, SUM(m.valore) AS totale" +
                            " ORDER BY totale DESC");

            for (Record rec : rs1) {
                System.out.printf("  %-30s  n=%-4s  totale=%.2f%n",
                        rec.getValue("causale"),
                        rec.getValue("num_movimenti"),
                        toDouble(rec.getValue("totale")));
            }

            System.out.println("-- Movimenti per conto e anno --");

            ResultSet rs2 = graph.query(
                    "MATCH (m:Movimento)-[:SU_CONTO]->(ct:Conto)" +
                            " RETURN ct.nome AS conto, m.anno AS anno, SUM(m.valore) AS totale" +
                            " ORDER BY ct.nome, m.anno");

            for (Record rec : rs2) {
                System.out.printf("  %-30s  %s  totale=%.2f%n",
                        rec.getValue("conto"),
                        rec.getValue("anno"),
                        toDouble(rec.getValue("totale")));
            }

            System.out.println("-- Conti collegati alla causale 'Stipendio' --");

            ResultSet rs3 = graph.query(
                    "MATCH (m:Movimento)-[:HA_CAUSALE]->(c:Causale {nome: 'Stipendio'})," +
                            "      (m)-[:SU_CONTO]->(ct:Conto)" +
                            " RETURN ct.nome AS conto, ct.iban AS iban, SUM(m.valore) AS totale");

            Iterator<Record> it3 = rs3.iterator();

            if (!it3.hasNext()) {
                System.out.println("  (nessun risultato - adatta il nome della causale)");
            } else {
                while (it3.hasNext()) {
                    Record rec = it3.next();
                    System.out.printf("  %-30s  %s  totale=%.2f%n",
                            rec.getValue("conto"),
                            rec.getValue("iban"),
                            toDouble(rec.getValue("totale")));
                }
            }
        }
    }

    private static String escape(String s) {
        return s == null ? "" : s.replace("'", "\'");
    }

    private static double toDouble(Object val) {
        if (val instanceof Number) {
            return ((Number) val).doubleValue();
        }
        try {
            return Double.parseDouble(val.toString());
        } catch (Exception e) {
            return 0;
        }
    }
}

Visto che sto facendo un pò di test, ogni volta ripulisco il grafo; vedete voi se farlo o no.

Enjoy!