Argthtjtr

I have a Spark structured streaming which reads UI events from a couple of busy Kafka topics.
Current flow is like this:

1. Spark stream reads the Kafka offsets


2. For every offset it goes to the database and maps one of values coming from the topic to another value.


3. Aggregates the data


4. Writes the data to the same database.

Facing this issue where after running 10-12 hours it throws too many db connection open error. It only does it for Step 2.

Using Aerospike Database for this Spark job.
Is there a way to optimize this flow? Is there calls to database can be reduced?

Read data:

sparkSession.readStream()

          .format("kafka")

          .option("kafka.bootstrap.servers", kafkaBootstrapServersString)

          .option("subscribe", newTopic)

          .option("startingOffsets", "latest")

          .option("enable.auto.commit", false)

          .option("failOnDataLoss", false)

          .load();

Map a value from database and aggregate data:

dataset

        .map(

            new MapFunction<Row, Row>() {

              @Override

              public Row call(Row row) throws Exception {

                objects[1] = aerospikeDao.getSomeValueFromCode(row.getAs("code"));



                return new GenericRowWithSchema(objects, eventSpecificStructType);

              }

            },

            RowEncoder.apply(eventSpecificStructType)

        )

        .withWatermark("timestamp", "30 seconds")

        .select(

            col("timestamp"),

            col("platform"),

            col("some_value")

        )

        .groupBy(

            functions.window(col("timestamp"), "30 seconds"),

            col("platform"),

            col("some_value")

        )

        .agg(

            count(lit(1)).as("count")

        );

Writing to the database:

aggregatedDataset

        .writeStream()

        .option("startingOffsets", "earliest")

        .outputMode(OutputMode.Append())

        .foreach(sink)

        .trigger(Trigger.ProcessingTime("30 seconds"))

        .start();

DAO:

    public AerospikeClient connect() {

        if (aerospikeClient == null || !aerospikeClient.isConnected()) {

          setAerospikeClient();

        }

        return this.aerospikeClient;

      }



 public void close() {

        if (aerospikeClient != null && aerospikeClient.isConnected()) {

            aerospikeClient.close();

        }

    }

public String getSomeValueFromCode(String code) {

    connet();

    Record record = aerospikeClient.get(Policy, key, "SomeValue");

    close();

    return channel;

  }