in my code, is the config of ExecutionEnv alright?
> On Aug 11, 2016, at 8:47 PM, Dong-iL, Kim <kim.s...@gmail.com> wrote:
>
>
> my code and log is as below.
>
>
> val getExecuteEnv: StreamExecutionEnvironment = {
> val env =
> StreamExecutionEnvironment.getExecutionEnvironment.enableCheckpointing(10000)
> env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
>
> env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE)
> env.getCheckpointConfig.setCheckpointTimeout(60000)
> env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
> env.setRestartStrategy(RestartStrategies.fixedDelayRestart(30, 30000))
> env
> }
>
> def transform(target: DataStream[(String, String, String, String, Long)]):
> DataStream[WinLossBase] =
> target.keyBy(_._3).flatMap(new StateOperator)
>
> def main(args: Array[String]) {
> val env = getExecuteEnv
> val source: DataStream[String] =
> extractFromKafka(env).name("KafkaSource")
> val json = deserializeToJsonObj(source).name("ConvertToJson")
> val target: DataStream[(String, String, String, String, Long)] =
> preTransform(json)
> val result: DataStream[WinLossBase] =
> transform(target).name("ToKeyedStream”)
> …
> }
>
> class StateOperator extends RichFlatMapFunction[(String, String, String,
> String, Long), WinLossBase] {
> var playerState: ValueState[util.Map[String, PotPlayer]] = _
> var handState: ValueState[HandHistoryInfo] = _
>
> override def open(param: Configuration): Unit = {
> val playerValueStateDescriptor = new
> ValueStateDescriptor[util.Map[String, PotPlayer]]("winloss",
> classOf[util.Map[String, PotPlayer]], Maps.newHashMap[String,
> PotPlayer]())
> playerState =
> getRuntimeContext.getState(playerValueStateDescriptor)
> handState = getRuntimeContext.getState(new
> ValueStateDescriptor("handinfo", classOf[HandHistoryInfo], null))
> }
>
> override def flatMap(in: (String, String, String, String, Long), out:
> Collector[WinLossBase]): Unit = {
> in._2 match {
> case "GameStartHistory" =>
> val players = playerState.value()
> val obj = _convertJsonToRecord(in._4,
> classOf[GameStartHistoryRecord])
> val record = obj.asInstanceOf[GameStartHistoryRecord]
> val handHistoryInfo: HandHistoryInfo =
> _setUpHandHistoryInfo(record)
> if (LOG.isInfoEnabled())
> LOG.info("hand start {}", if (handHistoryInfo != null)
> handHistoryInfo.handHistoryId else "NULL”)
> ….
> playerState.update(players)
> handState.update(handHistoryInfo)
> case "HoleCardHistory" =>
> val players = playerState.value()
> if (players != null) {
> ...
> playerState.update(players)
> } else LOG.warn("there is no player[hole card]. {}", in._4)
> case "PlayerStateHistory" =>
> val players = playerState.value()
> if (players != null) {
> ….
> playerState.update(players)
> } else LOG.warn("there is no player[player state]. {}",
> in._4)
> case "CommCardHistory" =>
> val handHistoryInfo = handState.value()
> val commCardHistory: CommCardHistory =
> commCardState.value()
> if (handHistoryInfo != null) {
> ...
> handState.update(handHistoryInfo)
> commCardState.update(commCardHistory)
> } else LOG.warn("there is no handhistory info[comm card].
> {}", in._4)
> case "PlayerActionHistory" =>
> val handHistoryInfo = handState.value()
> val players = playerState.value()
>
> if (handHistoryInfo != null) {
> ...
> } else LOG.warn("there is no handhistory info[player
> action]. {}", in._4)
> case "PotHistory" =>
> val players = playerState.value()
> val handHistoryInfo = handState.value()
> val commCardHistory: CommCardHistory =
> commCardState.value()
> if (handHistoryInfo != null && handHistoryInfo.playType ==
> PlayType.Cash && players != null && players.size > 1) {
> ...
> } else LOG.warn("there is no handhistory info[pot]. {}",
> in._4)
> case "GameEndHistory" =>
> val players = playerState.value()
> val handHistoryInfo = handState.value()
> ...
> if (LOG.isTraceEnabled()) LOG.trace("end {}",
> record.getHandHistoryId)
> playerState.clear()
> handState.clear()
> case _ =>
> }
> }
>
> —— log ——
> 2016-08-11 11:44:53.258 [ToKeyedStream -> (Map -> Sink: Pot to HBase, Map,
> Map -> Sink: winloss to HBase, Map -> Sink: winloss for opponent to HBase)
> (3/4)] INFO com.nsuslab.denma.stream.winloss.flow.Main$ - hand start
> 5769392597641628595
>
> 2016-08-11 11:45:07.555 [ToKeyedStream -> (Map -> Sink: Pot to HBase, Map,
> Map -> Sink: winloss to HBase, Map -> Sink: winloss for opponent to HBase)
> (3/4)] WARN com.nsuslab.denma.stream.winloss.flow.Main$ - there is no
> handhistory info[pot].
>
>> On Aug 11, 2016, at 7:01 PM, Ufuk Celebi <u...@apache.org> wrote:
>>
>> What do you mean with lost exactly?
>>
>> You call value() and it returns a value (!= null/defaultValue) and you
>> call it again and it returns null/defaultValue for the same key with
>> no update in between?
>>
>> On Thu, Aug 11, 2016 at 11:59 AM, Kostas Kloudas
>> <k.klou...@data-artisans.com> wrote:
>>> Hello,
>>>
>>> Could you share the code of the job you are running?
>>> With only this information I am afraid we cannot help much.
>>>
>>> Thanks,
>>> Kostas
>>>
>>>> On Aug 11, 2016, at 11:55 AM, Dong-iL, Kim <kim.s...@gmail.com> wrote:
>>>>
>>>> Hi.
>>>> I’m using flink 1.0.3 on aws EMR.
>>>> sporadically value of ValueState is lost.
>>>> what is starting point for solving this problem.
>>>> Thank you.
>>>
>
