(DEPRECATED) Apache Flink User Mailing List archive.

mapreduce.HadoopOutputFormat config value issue

classic Classic list List threaded Threaded
♦
♦
 2 messages Options
Shannon Carey
Reply | Threaded
Open this post in threaded view
♦
♦
|

mapreduce.HadoopOutputFormat config value issue

Shannon Carey
In Flink 1.1.1, I am seeing what looks like a serialization issue of org.apache.hadoop.conf.Configuration or when used with org.apache.flink.api.scala.hadoop.mapreduce.HadoopOutputFormat. When I use the mapred.HadoopOutputFormat version, it works just fine.

Specifically, the job fails because "java.lang.UnsupportedOperationException: You must set the ColumnFamily schema using setColumnFamilySchema." I am definitely setting that property, and it appears to be getting serialized, but when the config deserializes the setting is gone. Anybody have any ideas? In the meantime, I will continue using the "mapred" package.

Stack trace:
java.lang.UnsupportedOperationException: You must set the ColumnFamily schema using setColumnFamilySchema.
at org.apache.cassandra.hadoop.cql3.CqlBulkOutputFormat.getColumnFamilySchema(CqlBulkOutputFormat.java:184)
at org.apache.cassandra.hadoop.cql3.CqlBulkRecordWriter.setConfigs(CqlBulkRecordWriter.java:94)
at org.apache.cassandra.hadoop.cql3.CqlBulkRecordWriter.<init>(CqlBulkRecordWriter.java:74)
at org.apache.cassandra.hadoop.cql3.CqlBulkOutputFormat.getRecordWriter(CqlBulkOutputFormat.java:86)
at org.apache.cassandra.hadoop.cql3.CqlBulkOutputFormat.getRecordWriter(CqlBulkOutputFormat.java:52)
at org.apache.flink.api.java.hadoop.mapreduce.HadoopOutputFormatBase.open(HadoopOutputFormatBase.java:146)
at org.apache.flink.runtime.operators.DataSinkTask.invoke(DataSinkTask.java:176)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)
at java.lang.Thread.run(Thread.java:745)


Code that works:

    val insertStmt = s"INSERT INTO $keyspace.$colFamily (user_id, updated_time, value) VALUES (?, ?, ?)"
    val config = new JobConf()

    ConfigHelper.setOutputInitialAddress(config, initialOutputAddress.getHostAddress)

    CqlBulkOutputFormat.setColumnFamilySchema(config, colFamily, cqlTableSchema)
    CqlBulkOutputFormat.setColumnFamilyInsertStatement(config, colFamily, insertStmt)
    CqlBulkOutputFormat.setDeleteSourceOnSuccess(config, true)
    ConfigHelper.setOutputColumnFamily(config,
      keyspace,
      colFamily)
    ConfigHelper.setOutputPartitioner(config, partitionerClass)

    val outputFormat = new mapred.HadoopOutputFormat[Object, java.util.List[ByteBuffer]](
      new CqlBulkOutputFormat,
      config)

Code that doesn't work:

    val insertStmt = s"INSERT INTO $keyspace.$colFamily (user_id, updated_time, value) VALUES (?, ?, ?)"
    val config = new Configuration()

    ConfigHelper.setOutputInitialAddress(config, initialOutputAddress.getHostAddress)

    CqlBulkOutputFormat.setColumnFamilySchema(config, colFamily, cqlTableSchema)
    CqlBulkOutputFormat.setColumnFamilyInsertStatement(config, colFamily, insertStmt)
    CqlBulkOutputFormat.setDeleteSourceOnSuccess(config, true)
    ConfigHelper.setOutputColumnFamily(config,
      keyspace,
      colFamily)
    ConfigHelper.setOutputPartitioner(config, partitionerClass)

    val hadoopJob: Job = Job.getInstance(config)

    val outputFormat = new mapreduce.HadoopOutputFormat[Object, java.util.List[ByteBuffer]](
      new CqlBulkOutputFormat,
      hadoopJob)
Fabian Hueske-2
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: mapreduce.HadoopOutputFormat config value issue

Fabian Hueske-2
Hi Shannon,

I tried to reproduce the problem in a unit test without success.
My test configures a HadoopOutputFormat object, serializes and deserializes it, cally open, and verifies that a configured String property is present in the getRecordWriter() method.

Next I would try to reproduce the error with Cassandra. Which version are you using?
Can you also open a JIRA issue for this bug?

Thanks, Fabian

2016-10-11 23:40 GMT+02:00 Shannon Carey <[hidden email]>:
In Flink 1.1.1, I am seeing what looks like a serialization issue of org.apache.hadoop.conf.Configuration or when used with org.apache.flink.api.scala.hadoop.mapreduce.HadoopOutputFormat. When I use the mapred.HadoopOutputFormat version, it works just fine.

Specifically, the job fails because "java.lang.UnsupportedOperationException: You must set the ColumnFamily schema using setColumnFamilySchema." I am definitely setting that property, and it appears to be getting serialized, but when the config deserializes the setting is gone. Anybody have any ideas? In the meantime, I will continue using the "mapred" package.

Stack trace:
java.lang.UnsupportedOperationException: You must set the ColumnFamily schema using setColumnFamilySchema.
at org.apache.cassandra.hadoop.cql3.CqlBulkOutputFormat.getColumnFamilySchema(CqlBulkOutputFormat.java:184)
at org.apache.cassandra.hadoop.cql3.CqlBulkRecordWriter.setConfigs(CqlBulkRecordWriter.java:94)
at org.apache.cassandra.hadoop.cql3.CqlBulkRecordWriter.<init>(CqlBulkRecordWriter.java:74)
at org.apache.cassandra.hadoop.cql3.CqlBulkOutputFormat.getRecordWriter(CqlBulkOutputFormat.java:86)
at org.apache.cassandra.hadoop.cql3.CqlBulkOutputFormat.getRecordWriter(CqlBulkOutputFormat.java:52)
at org.apache.flink.api.java.hadoop.mapreduce.HadoopOutputFormatBase.open(HadoopOutputFormatBase.java:146)
at org.apache.flink.runtime.operators.DataSinkTask.invoke(DataSinkTask.java:176)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)
at java.lang.Thread.run(Thread.java:745)


Code that works:

    val insertStmt = s"INSERT INTO $keyspace.$colFamily (user_id, updated_time, value) VALUES (?, ?, ?)"
    val config = new JobConf()

    ConfigHelper.setOutputInitialAddress(config, initialOutputAddress.getHostAddress)

    CqlBulkOutputFormat.setColumnFamilySchema(config, colFamily, cqlTableSchema)
    CqlBulkOutputFormat.setColumnFamilyInsertStatement(config, colFamily, insertStmt)
    CqlBulkOutputFormat.setDeleteSourceOnSuccess(config, true)
    ConfigHelper.setOutputColumnFamily(config,
      keyspace,
      colFamily)
    ConfigHelper.setOutputPartitioner(config, partitionerClass)

    val outputFormat = new mapred.HadoopOutputFormat[Object, java.util.List[ByteBuffer]](
      new CqlBulkOutputFormat,
      config)

Code that doesn't work:

    val insertStmt = s"INSERT INTO $keyspace.$colFamily (user_id, updated_time, value) VALUES (?, ?, ?)"
    val config = new Configuration()

    ConfigHelper.setOutputInitialAddress(config, initialOutputAddress.getHostAddress)

    CqlBulkOutputFormat.setColumnFamilySchema(config, colFamily, cqlTableSchema)
    CqlBulkOutputFormat.setColumnFamilyInsertStatement(config, colFamily, insertStmt)
    CqlBulkOutputFormat.setDeleteSourceOnSuccess(config, true)
    ConfigHelper.setOutputColumnFamily(config,
      keyspace,
      colFamily)
    ConfigHelper.setOutputPartitioner(config, partitionerClass)

    val hadoopJob: Job = Job.getInstance(config)

    val outputFormat = new mapreduce.HadoopOutputFormat[Object, java.util.List[ByteBuffer]](
      new CqlBulkOutputFormat,
      hadoopJob)
Free forum by Nabble Edit this page