9567 - A Guide to Apache Paimon Java API

persistence-modules/apache-paimon/pom.xml

@@ -0,0 +1,58 @@
+<?xml version="1.0"?>
+<project
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"
+  xmlns="http://maven.apache.org/POM/4.0.0"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>com.baeldung</groupId>
+    <artifactId>persistence-modules</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+  </parent>
+  <groupId>com.baeldung</groupId>
+  <artifactId>apache-paimon</artifactId>
+  <version>1.0-SNAPSHOT</version>
+  <name>apache-paimon</name>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.paimon</groupId>
+      <artifactId>paimon-bundle</artifactId>
+      <version>${paimon.version}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-api</artifactId>
+        </exclusion>
+      </exclusions>
+
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-client-runtime</artifactId>
+      <scope>runtime</scope>
+      <version>${hadoop.version}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-api</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.xerial.snappy</groupId>
+          <artifactId>snappy-java</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>com.google.code.findbugs</groupId>
+          <artifactId>jsr305</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+  </dependencies>
+
+  <properties>
+    <paimon.version>1.4.1</paimon.version>
+    <hadoop.version>3.4.3</hadoop.version>
+    <java.version>17</java.version>
+  </properties>
+</project>

persistence-modules/apache-paimon/src/main/java/com/baeldung/paimon/Metric.java

@@ -0,0 +1,87 @@
+package com.baeldung.paimon;
+
+public class Metric {
+    private String deviceId;
+    private String metricsName;
+    private double metricsValue;
+    private String source;
+    private String state = "active";
+    private String createTime;
+    private String createdBy;
+
+    public Metric() {
+    }
+
+    public Metric(String deviceId, String metricsName, double metricsValue, String source, String createTime, String createdBy, String state) {
+        this.deviceId = deviceId;
+        this.metricsName = metricsName;
+        this.metricsValue = metricsValue;
+        this.source = source;
+        this.createTime = createTime;
+        this.createdBy = createdBy;
+        this.state = state;
+    }
+
+    public String getCreatedBy() {
+        return createdBy;
+    }
+
+    public void setCreatedBy(String createdBy) {
+        this.createdBy = createdBy;
+    }
+
+
+    public String getDeviceId() {
+        return deviceId;
+    }
+
+    public void setDeviceId(String deviceId) {
+        this.deviceId = deviceId;
+    }
+
+    public String getMetricsName() {
+        return metricsName;
+    }
+
+    public void setMetricsName(String metricsName) {
+        this.metricsName = metricsName;
+    }
+
+    public double getMetricsValue() {
+        return metricsValue;
+    }
+
+    public void setMetricsValue(double metricsValue) {
+        this.metricsValue = metricsValue;
+    }
+
+    public String getSource() {
+        return source;
+    }
+
+    public void setSource(String source) {
+        this.source = source;
+    }
+
+    public String getCreateTime() {
+        return createTime;
+    }
+
+    public void setCreateTime(String createTime) {
+        this.createTime = createTime;
+    }
+
+    public String getState() {
+        return state;
+    }
+
+    public void setState(String state) {
+        this.state = state;
+    }
+
+    @Override
+    public String toString() {
+        return "Metric [deviceId=" + deviceId + ", metricsName=" + metricsName + ", metricsValue="
+                + metricsValue + ", source=" + source + ", createTime=" + createTime + ", state=" + state + "]";
+    }
+}

persistence-modules/apache-paimon/src/main/java/com/baeldung/paimon/MetricReader.java

@@ -0,0 +1,53 @@
+package com.baeldung.paimon;
+
+import java.io.BufferedReader;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MetricReader {
+
+    Logger logger = LoggerFactory.getLogger(MetricReader.class);
+
+    List<Metric> readMetrics() {
+        ClassLoader classLoader = getClass().getClassLoader();
+        InputStream inputStream = classLoader.getResourceAsStream("metrics.out");
+
+        BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
+        String line;
+        String CREATED_BY = "admin";
+        List<Metric> metrics = new ArrayList<>();
+        boolean isFirstLine = true;
+        try {
+            while ((line = reader.readLine()) != null) {
+                if (isFirstLine) {
+                    isFirstLine = false;
+                    continue;
+                }
+                String[] parts = line.split(",");
+                if (parts.length == 5) {
+                    String deviceId = parts[0];
+                    String metricsName = parts[1];
+
+
+                   String createTime = parts[2];
+
+                    long metricsValue = (long) Double.parseDouble(parts[3]);
+                    String source = parts[4];
+                    logger.info("Read metric: deviceId={}, metricsName={}, createTime={}, metricsValue={}, source={}",
+                            deviceId, metricsName, createTime, metricsValue, source);
+
+                    Metric metric = new Metric(deviceId, metricsName, metricsValue, source, createTime, CREATED_BY, "active");
+                    metrics.add(metric);
+                }
+            }
+        } catch (Exception e) {
+            logger.error("Error reading metrics", e);
+        }
+
+        return metrics;
+    }   
+}

persistence-modules/apache-paimon/src/main/java/com/baeldung/paimon/PaimonDatabaseManager.java

@@ -0,0 +1,36 @@
+package com.baeldung.paimon;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.types.DataTypes;
+
+public class PaimonDatabaseManager {
+    public static Catalog createCatalog(String warehousePath) {
+        CatalogContext context = CatalogContext.create(new Path(warehousePath));
+        return CatalogFactory.createCatalog(context);
+    }
+
+    public static Identifier createTable(Catalog catalog) throws Exception {
+
+        Schema schema = Schema.newBuilder()
+            .column("device_id", DataTypes.STRING())
+            .column("metrics_name", DataTypes.STRING())
+            .column("metrics_value", DataTypes.DOUBLE())
+            .column("source", DataTypes.STRING())
+            .column("create_time", DataTypes.TIMESTAMP(3))
+            .column("created_by", DataTypes.STRING())
+            .column("state", DataTypes.STRING())
+            .primaryKey("device_id", "metrics_name", "create_time")  
+            .build();
+
+        Identifier tableId = Identifier.create("metric_db", "metrics");
+
+        catalog.createDatabase("metric_db", false);  
+        catalog.createTable(tableId, schema, false);
+        return tableId;
+    }
+}