It seems that it's related with your development environment settings.
On Sun, Jul 29, 2018 at 4:39 PM Mich Talebzadeh <[email protected]>
wrote:
> Hi,
>
> I have written a simple test program as below
>
> import java.util.Properties
> import java.util.Arrays
> import org.apache.flink.api.common.functions.MapFunction
> import org.apache.flink.api.java.utils.ParameterTool
> import org.apache.flink.streaming.api.datastream.DataStream
> import
> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
> import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011
> import org.apache.flink.streaming.util.serialization.SimpleStringSchema
> import org.apache.flink.streaming.util.serialization.DeserializationSchema
> import org.apache.flink.streaming.util.serialization.SimpleStringSchema
> import org.apache.kafka.clients.consumer.ConsumerConfig
> import org.apache.kafka.clients.consumer.ConsumerRecord
> import org.apache.kafka.clients.consumer.ConsumerRecords
> import org.apache.kafka.clients.consumer.KafkaConsumer
> import org.apache.flink.core.fs.FileSystem
> import org.apache.flink.streaming.api
> import org.apache.flink.streaming.api.TimeCharacteristic
> import org.apache.flink.streaming.api.datastream.DataStream
> import
> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
> import
> org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
> import org.apache.flink.streaming.api.watermark.Watermark
> import org.apache.flink.streaming.api.windowing.time.Time
> import org.apache.flink.util.Collector
> import
> org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
> import org.apache.flink.streaming.api.watermark.Watermark
> import org.apache.flink.streaming.api.functions.source.SourceFunction
> import org.apache.flink.streaming.connectors.kafka
> object md_streaming
> {
> private var zookeeperUrl = "rhes75:2181"
> private var requestConsumerId = null
> private var impressionConsumerId = null
> private var clickConsumerId = null
> private var conversionConsumerId = null
> private var requestTopicName = null
> private var impressionTopicName = null
> private var clickTopicName = null
> private var conversionTopicName = null
> private var requestThreads = 0
> private var impressionThreads = 0
> private var clickThreads = 0
> private var conversionThreads = 0
> private var flinkAppName = "md_streaming"
> private var bootstrapServers = "rhes75:9092, rhes75:9093, rhes75:9094"
> private var schemaRegistryURL = "http://rhes75:8081";
> private var zookeeperConnect = "rhes75:2181"
> private var zookeeperConnectionTimeoutMs = "10000"
> private var rebalanceBackoffMS = "15000"
> private var zookeeperSessionTimeOutMs = "15000"
> private var autoCommitIntervalMS = "12000"
> private var topicsValue = "final"
> private var memorySet = "F"
> private var enableHiveSupport = null
> private var enableHiveSupportValue = "true"
> private var sparkStreamingReceiverMaxRateValue = "0"
> private var checkpointdir = "/checkpoint"
> private var hbaseHost = "rhes75"
> private var zookeeperHost = "rhes564"
> private var zooKeeperClientPort = "2181"
> private var writeDirectory = "hdfs://rhes75:9000/tmp/flink/"
> private var fileName = "md_streaming.txt"
> private val maxServingDelay = 60
> private val servingSpeedFactor = 600f
> private var batchInterval = 2
> private val countWindowLength = 4 // window size in sec
> private val countWindowFrequency = 2 // window trigger interval in sec
> private val earlyCountThreshold = 50
> private val writeToElasticsearch = false // set to true to write results
> to Elasticsearch
> private val elasticsearchHost = "" // look-up hostname in Elasticsearch
> log output
> private val elasticsearchPort = 9300
>
> def main(args: Array[String])
> {
> val env = StreamExecutionEnvironment.getExecutionEnvironment
> env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
>
> val properties = new Properties()
> properties.setProperty("bootstrap.servers", bootstrapServers)
> properties.setProperty("zookeeper.connect", zookeeperConnect)
> properties.setProperty("group.id", flinkAppName)
> properties.setProperty("auto.offset.reset", "latest")
>
> val myConsumer = env.addSource(new
> FlinkKafkaConsumer011[String](topicsValue, new SimpleStringSchema(),
> properties))
>
> myConsumer.assignTimestampsAndWatermarks(new CustomWatermarkEmitter())
> env.addSource(myConsumer).print()
> //val sink = kafkaSource.writeAsText(writeDirectory+fileName,
> FileSystem.WriteMode.OVERWRITE)
> //env.execute("Flink Kafka Example writing to
> "+writeDirectory+fileName)
> env.execute("Flink simple output")
> }
> }
>
> However, when compiling I am getting the following errors
>
> [error]
> /home/hduser/dba/bin/flink/md_streaming/src/main/scala/myPackage/md_streaming.scala:98:
> not found: type CustomWatermarkEmitter
> [error] myConsumer.assignTimestampsAndWatermarks(new
> CustomWatermarkEmitter())
> [error] ^
> [error]
> /home/hduser/dba/bin/flink/md_streaming/src/main/scala/myPackage/md_streaming.scala:99:
> type mismatch;
> [error] found :
> org.apache.flink.streaming.api.datastream.DataStreamSource[String]
> [error] required:
> org.apache.flink.streaming.api.functions.source.SourceFunction[?]
> [error] env.addSource(myConsumer).print()
> [error] ^
> [error] two errors found
> [error] (compile:compileIncremental) Compilation failed
> [error] Total time: 3 s, completed Jul 29, 2018 9:31:05 AM
> Completed compiling
> Sun Jul 29 09:31:05 BST 2018 , Running in **Standalone mode**
> Could not build the program from JAR file.
>
> I don't see why it is failing. Appreciate any suggestions.
>
> Regards,
>
> Dr Mich Talebzadeh
>
>
>
> LinkedIn *
> https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw
> <https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw>*
>
>
>
> http://talebzadehmich.wordpress.com
>
>
> *Disclaimer:* Use it at your own risk. Any and all responsibility for any
> loss, damage or destruction of data or any other property which may arise
> from relying on this email's technical content is explicitly disclaimed.
> The author will in no case be liable for any monetary damages arising from
> such loss, damage or destruction.
>
>
>
--
Liu, Renjie
Software Engineer, MVAD
