And of course the moment I click "send" I find that: 😂
If you use Scala 2.11 and dependency version 1.8.0 in your Beam projects
pom.xml it *does* work:
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-statebackend-rocksdb_2.11</artifactId>
<version>1.8.0</version>
</dependency>
However, if you want to use 1.8.1 - it *does not*.
I still found it confusing, as I am using the official Flink Docker images
which are currently at version 1.8.1. It would have helped me if Beam would
bundle the statebackend dependency (as already mentioned Beam allows the
user to set a state backend via parameters of the Flink Runner).
On Tue, Aug 6, 2019 at 4:35 PM Kaymak, Tobias <[email protected]>
wrote:
> Hello,
>
> Flink requires in version 1.8, that if one wants to use RocksDB as a state
> backend, that dependency has to be added to the pom.xml file. [0]
>
> My cluster stopped working with RocksDB so I did added this dependency to
> the pom.xml of my Beam project (I've tried 1.8.1 and 1.8.0):
>
> <dependency>
> <groupId>org.apache.flink</groupId>
> <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
> <version>1.8.0</version>
> </dependency>
>
> I also tried to instead add the flink-statebackend-rocksdb_2.11-1.8.0.jar
> to the lib directory of the Flink cluster instead (TaskManagers and
> JobManager) in all cases I get this error:
>
> 2019-08-06 14:14:15,670 ERROR
> org.apache.flink.streaming.runtime.tasks.StreamTask - Error
> during disposal of stream operator
> java.lang.NoSuchMethodError:
> org.rocksdb.ColumnFamilyHandle.getDescriptor()Lorg/rocksdb/ColumnFamilyDescriptor;
> at
> org.apache.flink.contrib.streaming.state.RocksDBOperationUtils.addColumnFamilyOptionsToCloseLater(RocksDBOperationUtils.java:160)
> at
> org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend.dispose(RocksDBKeyedStateBackend.java:331)
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.dispose(AbstractStreamOperator.java:362)
> at
> org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator.dispose(DoFnOperator.java:470)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.disposeAllOperators(StreamTask.java:477)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:378)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)
> at java.lang.Thread.run(Thread.java:748)
>
> This looks like a version mismatch to me, but I don't know how to solve it
> - could Beam maybe include the dependency for the RocksDB backend for Flink
> 1.8 or higher, as it allows to set this value via parameters for the Flink
> Runner? [1]
>
>
> [0]
> https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/state/state_backends.html#setting-the-per-job-state-backend
> [1]
> https://beam.apache.org/documentation/runners/flink/#pipeline-options-for-the-flink-runner
>
