OrientDB

The Apache Pekko Connectors OrientDB connector provides Apache Pekko Stream sources and sinks for OrientDB.

Project Info: Apache Pekko Connectors OrientDB
Artifact	org.apache.pekko pekko-connectors-orientdb 1.1.0 Snapshots are available
JDK versions	OpenJDK 8 OpenJDK 11 OpenJDK 17 OpenJDK 21
Scala versions	2.13.15, 2.12.20, 3.3.4
JPMS module name	pekko.stream.connectors.orientdb
License	Apache-2.0
API documentation	API (Scaladoc)
Forums	Github Discussions
Release notes	GitHub releases
Issues	Github issues
Sources	https://github.com/apache/pekko-connectors

Artifacts

sbt

val PekkoVersion = "1.1.3"
libraryDependencies ++= Seq(
  "org.apache.pekko" %% "pekko-connectors-orientdb" % "1.1.0",
  "org.apache.pekko" %% "pekko-stream" % PekkoVersion
)

Maven

<properties>
  <pekko.version>1.1.3</pekko.version>
  <scala.binary.version>2.13</scala.binary.version>
</properties>
<dependencies>
  <dependency>
    <groupId>org.apache.pekko</groupId>
    <artifactId>pekko-connectors-orientdb_${scala.binary.version}</artifactId>
    <version>1.1.0</version>
  </dependency>
  <dependency>
    <groupId>org.apache.pekko</groupId>
    <artifactId>pekko-stream_${scala.binary.version}</artifactId>
    <version>${pekko.version}</version>
  </dependency>
</dependencies>

Gradle

def versions = [
  PekkoVersion: "1.1.3",
  ScalaBinary: "2.13"
]
dependencies {
  implementation "org.apache.pekko:pekko-connectors-orientdb_${versions.ScalaBinary}:1.1.0"
  implementation "org.apache.pekko:pekko-stream_${versions.ScalaBinary}:${versions.PekkoVersion}"
}

The table below shows direct dependencies of this module and the second tab shows all libraries it depends on transitively.

Database connection

Sources, Flows and Sinks provided by this connector need a OPartitionedDatabasePool to access to OrientDB. It is your responsibility to close the database connection eg. at actor system termination. This API has become deprecated in OrientDB, please suggest a Pull Request to use the latest APIs instead.

Scala

copysourceimport com.orientechnologies.orient.core.db.OPartitionedDatabasePool

val url = "remote:127.0.0.1:2424/"
val dbName = "GratefulDeadConcertsScala"
val dbUrl = s"$url$dbName"
val username = "root"
val password = "root"

  val oDatabase: OPartitionedDatabasePool =
    new OPartitionedDatabasePool(dbUrl, username, password, Runtime.getRuntime.availableProcessors(), 10)

  system.registerOnTermination(() -> oDatabase.close())

Java

copysourceimport com.orientechnologies.orient.core.db.OPartitionedDatabasePool;

private static String url = "remote:127.0.0.1:2424/";
private static String dbName = "GratefulDeadConcertsJava";
private static String dbUrl = url + dbName;
private static String username = "root";
private static String password = "root";

  oDatabase =
      new OPartitionedDatabasePool(
          dbUrl, username, password, Runtime.getRuntime().availableProcessors(), 10);

  system.registerOnTermination(() -> oDatabase.close());

Reading ODocument from OrientDB

Now we can stream messages which contain OrientDB’s ODocument (in Scala or Java) from or to OrientDB by providing the ODatabaseDocumentTx to the OrientDbSourceOrientDbSource.

Scala

copysourceval result: Future[immutable.Seq[String]] = OrientDbSource(
  sink4,
  OrientDbSourceSettings(oDatabase)).map { (message: OrientDbReadResult[ODocument]) =>
  message.oDocument.field[String]("book_title")
}
  .runWith(Sink.seq)

Java

copysourceCompletionStage<List<String>> result =
    OrientDbSource.create(sinkClass1, OrientDbSourceSettings.create(oDatabase))
        .map(m -> m.oDocument().<String>field("book_title"))
        .runWith(Sink.seq(), system);

Typed messages

Also, it’s possible to stream messages which contains any classes.

Java

copysourcepublic static class source1 {

  private String book_title;

  public void setBook_title(String book_title) {
    this.book_title = book_title;
  }

  public String getBook_title() {
    return book_title;
  }
}

public static class sink2 {

  private String book_title;

  public void setBook_title(String book_title) {
    this.book_title = book_title;
  }

  public String getBook_title() {
    return book_title;
  }
}

Use OrientDbSource.typed and OrientDbSink.typed to create source and sink instead.

Scala

copysourceval streamCompletion: Future[Done] = OrientDbSource
  .typed(sourceClass, OrientDbSourceSettings(oDatabase), classOf[OrientDbTest.source1])
  .map { (m: OrientDbReadResult[OrientDbTest.source1]) =>
    val db: ODatabaseDocumentTx = oDatabase.acquire
    db.setDatabaseOwner(new OObjectDatabaseTx(db))
    ODatabaseRecordThreadLocal.instance.set(db)
    val sink: OrientDbTest.sink2 = new OrientDbTest.sink2
    sink.setBook_title(m.oDocument.getBook_title)
    OrientDbWriteMessage(sink)
  }
  .groupedWithin(10, 10.millis)
  .runWith(OrientDbSink.typed(sinkClass2, OrientDbWriteSettings.create(oDatabase), classOf[OrientDbTest.sink2]))

Java

copysourceCompletionStage<Done> f1 =
    OrientDbSource.typed(
            sourceClass, OrientDbSourceSettings.create(oDatabase), source1.class, null)
        .map(
            readResult -> {
              ODatabaseDocumentTx db = oDatabase.acquire();
              db.setDatabaseOwner(new OObjectDatabaseTx(db));
              ODatabaseRecordThreadLocal.instance().set(db);
              sink2 sink = new sink2();
              sink.setBook_title(readResult.oDocument().getBook_title());
              return OrientDbWriteMessage.create(sink);
            })
        .groupedWithin(10, Duration.ofMillis(10))
        .runWith(
            OrientDbSink.typed(
                sinkClass2, OrientDbWriteSettings.create(oDatabase), sink2.class),
            system);

Source configuration

We can configure the source by OrientDbSourceSettings.

Scala

copysource// re-iterating default values
val sourceSettings = OrientDbSourceSettings(oDatabase)
  .withSkip(0)
  .withLimit(10)

Java

copysource// re-iterating default values
OrientDbSourceSettings sourceSettings =
    OrientDbSourceSettings.create(oDatabase).withSkip(0).withLimit(10);

Parameter	Default	Description
skip	0	Rows skipped in the beginning of the result.
limit	10	Result items fetched per query.

Writing to OrientDB

You can also build flow stages. The API is similar to creating Sinks.

Scala

copysource
val f1 = OrientDbSource(
  sourceClass,
  OrientDbSourceSettings(oDatabase)).map { (message: OrientDbReadResult[ODocument]) =>
  OrientDbWriteMessage(message.oDocument)
}
  .groupedWithin(10, 50.millis)
  .via(
    OrientDbFlow.create(
      sink5,
      OrientDbWriteSettings(oDatabase)))
  .runWith(Sink.seq)

Java

copysourceCompletionStage<List<List<OrientDbWriteMessage<ODocument, NotUsed>>>> f1 =
    OrientDbSource.create(sourceClass, OrientDbSourceSettings.create(oDatabase), null)
        .map(m -> OrientDbWriteMessage.create(m.oDocument()))
        .groupedWithin(10, Duration.ofMillis(10))
        .via(OrientDbFlow.create(sink3, OrientDbWriteSettings.create(oDatabase)))
        .runWith(Sink.seq(), system);

Passing data through OrientDBFlow

When streaming documents from Kafka, you might want to commit to Kafka AFTER the document has been written to OrientDB.

Scala

copysource// We're going to pretend we got messages from kafka.
// After we've written them to oRIENTdb, we want
// to commit the offset to Kafka

case class KafkaOffset(offset: Int)
case class KafkaMessage(book: Book, offset: KafkaOffset)

val messagesFromKafka = List(
  KafkaMessage(Book("Book 1"), KafkaOffset(0)),
  KafkaMessage(Book("Book 2"), KafkaOffset(1)),
  KafkaMessage(Book("Book 3"), KafkaOffset(2)))

var committedOffsets = List[KafkaOffset]()

def commitToKafka(offset: KafkaOffset): Unit =
  committedOffsets = committedOffsets :+ offset

val f1 = Source(messagesFromKafka)
  .map { (kafkaMessage: KafkaMessage) =>
    val book = kafkaMessage.book
    val id = book.title
    println("title: " + book.title)

    OrientDbWriteMessage(new ODocument().field("book_title", id), kafkaMessage.offset)
  }
  .groupedWithin(10, 50.millis)
  .via(
    OrientDbFlow.createWithPassThrough(
      sink7,
      OrientDbWriteSettings(oDatabase)))
  .map { (messages: Seq[OrientDbWriteMessage[ODocument, KafkaOffset]]) =>
    messages.foreach { message =>
      commitToKafka(message.passThrough)
    }
  }
  .runWith(Sink.ignore)

Java

copysource// We're going to pretend we got messages from kafka.
// After we've written them to OrientDB, we want
// to commit the offset to Kafka

List<Integer> committedOffsets = new ArrayList<>();
List<messagesFromKafka> messagesFromKafkas =
    Arrays.asList(
        new messagesFromKafka("Akka Concurrency", new KafkaOffset(0)),
        new messagesFromKafka("Akka in Action", new KafkaOffset(1)),
        new messagesFromKafka("Effective Akka", new KafkaOffset(2)));

Consumer<KafkaOffset> commitToKafka =
    new Consumer<KafkaOffset>() {
      @Override
      public void accept(KafkaOffset kafkaOffset) {
        committedOffsets.add(kafkaOffset.getOffset());
      }
    };

Source.from(messagesFromKafkas)
    .map(
        kafkaMessage -> {
          String book_title = kafkaMessage.getBook_title();
          return OrientDbWriteMessage.create(
              new ODocument().field("book_title", book_title), kafkaMessage.kafkaOffset);
        })
    .groupedWithin(10, Duration.ofMillis(10))
    .via(OrientDbFlow.createWithPassThrough(sink6, OrientDbWriteSettings.create(oDatabase)))
    .map(
        messages -> {
          ODatabaseDocumentTx db = oDatabase.acquire();
          db.setDatabaseOwner(new OObjectDatabaseTx(db));
          ODatabaseRecordThreadLocal.instance().set(db);
          messages.stream().forEach(message -> commitToKafka.accept(message.passThrough()));
          return NotUsed.getInstance();
        })
    .runWith(Sink.seq(), system)
    .toCompletableFuture()
    .get(10, TimeUnit.SECONDS);