Yes Nix.
I can confirm that I am able to connect to InfluxDB from Flink(taskmanager).
URL="http://influxdb:8086";
And also, I tried with below from inside Flink(taskmanager) container
and that also worked.
curl influxdb:8086
Siva.
On Thu, Jan 30, 2025 at 4:50 PM Nikola Milutinovic
<n.milutino...@levi9.com> wrote:
>
> Hi.
>
>
>
> First of all, can you verify your Docker compose InfluxDB is working, is it
> accepting connections?
>
>
>
> Can you connect to it from outside Docker Compose?
>
>
>
> When connecting from Flink, I hope you are referencing the service name as a
> host, because that is how Docker Compose works. It is a common mistake to
> target “localhost”, which stays inside the container.
>
>
>
> Nix.
>
>
>
> From: Siva Krishna <sivasoe...@gmail.com>
> Date: Thursday, January 30, 2025 at 11:07 AM
> To: Nikola Milutinovic <n.milutino...@levi9.com>
> Cc: user@flink.apache.org <user@flink.apache.org>
> Subject: Re: Please help with Flink + InfluxDB
>
> Hi Nix,
> Thank you for your reply.
> I have small update. For trial/error, I have signed in InfluxDB cloud.
> This same code is working and data is being save in the cloud.
> So, I am guessing there might issue with my 'docker compose', which is
> why the data is not being saved in local(docker InfluxDB).
> Maybe I need to correct my volumes paths or something?
>
> Warm regards,
> Siva Rama Krishna.
>
> On Thu, Jan 30, 2025 at 2:05 PM Nikola Milutinovic
> <n.milutino...@levi9.com> wrote:
> >
> > Hi Siva.
> >
> >
> >
> > What is the InfluxDB URL you are using? I assume it is something like
> > “influxdb://influxdb:8086/…”. If it is “localhost:8086…” that would be a
> > problem.
> >
> >
> >
> > Do you see a connection on InfluxDB? Anything in the logs? Anything in the
> > logs of Job Mnager?
> >
> >
> >
> > Nix,
> >
> >
> >
> > From: Siva Krishna <sivasoe...@gmail.com>
> > Date: Thursday, January 30, 2025 at 5:12 AM
> > To: user@flink.apache.org <user@flink.apache.org>
> > Subject: Please help with Flink + InfluxDB
> >
> > Hi, I am trying to create a data streaming pipeline for real-time analytics.
> >
> > Kafka -> Flink -> InfluDB -> Visualize
> >
> > I am facing an issue writing the data to InfluxDB. I can see logs in
> > 'taskmanager' after submitting the Job. But, the data is not being
> > written to InfluxDB for some reason.
> > Please help, I am attaching sample docker setup and basic code.
> >
> > import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
> > import com.influxdb.client.InfluxDBClient
> > import com.influxdb.client.InfluxDBClientFactory
> > import com.influxdb.client.InfluxDBClientOptions
> > import com.influxdb.client.domain.WritePrecision
> > import com.influxdb.client.write.Point
> > import org.apache.flink.api.common.eventtime.Watermark
> > import org.apache.flink.api.connector.sink2.Sink
> > import org.apache.flink.api.connector.sink2.SinkWriter
> > import org.apache.flink.api.connector.sink2.WriterInitContext
> > import java.time.Instant
> >
> > class InfluxDBSink : Sink<String> {
> > override fun createWriter(p0: WriterInitContext?): SinkWriter<String> {
> > return InfluxDBSinkWriter()
> > }
> > }
> >
> > // Implement the SinkWriter that writes EventData to InfluxDB
> > class InfluxDBSinkWriter : SinkWriter<String> {
> > private val mapper = jacksonObjectMapper()
> >
> > // Initialize the InfluxDB client
> > private val influxDBClient = initializeInfluxDBClient()
> > // private val writeApi = influxDBClient.makeWriteApi()
> >
> > // Write data to InfluxDB
> > override fun write(element: String, context: SinkWriter.Context) {
> >
> > val event = mapper.readValue(element, EventData::class.java)
> > println("eventData: $event")
> >
> > try {
> > val connectable = influxDBClient.ping()
> > val ready = influxDBClient.ready()
> > val influxStr = influxDBClient.toString()
> > println("influxStr: $connectable -> $ready -> $influxStr")
> > } catch (e: Exception) {
> > println(e.message)
> > }
> >
> > val point = Point("dummy_events")
> > .addTag("source", "flink")
> > .addField("timestamp", event.timestamp)
> > .addField("value", event.value)
> > .time(Instant.now().toEpochMilli(), WritePrecision.MS)
> > // println("Writing point to InfluxDB: $point,
> > ${point.toLineProtocol()}")
> >
> > // writeApi.writePoint(point)
> > // writeApi.flush()
> >
> > influxDBClient.writeApiBlocking.writePoint(point)
> > // println("after influxDBClient writeApiBlocking")
> > }
> >
> > // This method will flush any pending writes (can be empty if no
> > additional batching needed)
> > override fun flush(endOfInput: Boolean) {
> > // if (endOfInput) writeApi.flush()
> > }
> >
> > // Handle any watermarks if required (optional, can be skipped if
> > not needed)
> > override fun writeWatermark(watermark: Watermark) {
> > }
> >
> > // Close the writer when done
> > override fun close() {
> > influxDBClient.close()
> > }
> > }
> >
> > fun initializeInfluxDBClient(): InfluxDBClient {
> > // Build the InfluxDBClientOptions with all required parameters
> > val influxDBOptions = InfluxDBClientOptions.builder()
> > .url(influxDBUrl) // InfluxDB URL
> > .authenticate(influxUserName, influxPassword.toCharArray())
> > // Username and password
> > .org(influxOrg) // Organization name
> > .bucket(influxBucket) // Bucket name
> > .authenticateToken(influxToken.toCharArray()) // Token for
> > authentication
> > .build()
> >
> > // Create the InfluxDB client using the built options
> > return InfluxDBClientFactory.create(influxDBOptions)
> > }
> >
> >
> > Main.kt
> > fun main() {
> > // Create KafkaSource with properties
> > val kafkaSource = KafkaSource.builder<String>()
> > .setBootstrapServers("kafka:9092")
> > .setTopics("dummy-events") // Kafka topic
> > .setGroupId("flink-group")
> > .setStartingOffsets(OffsetsInitializer.earliest()) // Start
> > from the latest message
> > .setValueOnlyDeserializer(SimpleStringSchema()) //
> > Deserialize string values
> > .setProperties(Properties().apply {
> > setProperty("auto.offset.reset", "earliest")
> > })
> > .build()
> >
> > // Create a WatermarkStrategy for event time processing
> > val watermarkStrategy = WatermarkStrategy.noWatermarks<String>()
> >
> > // Set up the execution environment
> > val env = StreamExecutionEnvironment.getExecutionEnvironment()
> >
> > // Add KafkaSource as the input stream, provide watermark strategy
> > and a source name
> > val stream = env.fromSource(
> > kafkaSource,
> > watermarkStrategy,
> > "KafkaSource"
> > )
> >
> > stream.sinkTo(InfluxDBSink())
> >
> > // Execute the job
> > env.execute("Flink Kafka Integration")
> > }
> >
> > But below curl is working.
> > curl --request POST
> > "http://localhost:8086/api/v2/write?org=org&bucket=sample_data&precision=ms";
> > \
> > --header "Authorization: Token
> > wvd-8VqKyz6YodOSyVPU17oSOobKqwJDHf0GtFzd9-1DuzFyF3l1xlDGt2R062CI8Jpg=="
> > \
> > --header "Content-Type: text/plain; charset=utf-8" \
> > --data-binary "dummy_events,source=flink value=42.0 1738142927000"
> >
> >
> >
> > Please check and reply is I am doing anything wrong.
> >
> > Thank you.
