flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Shannon Carey <sca...@expedia.com>
Subject mapreduce.HadoopOutputFormat config value issue
Date Tue, 11 Oct 2016 21:40:39 GMT
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)

Mime
View raw message