ogibayashi

---
hostPort: localhost:{{ item.jmx_port }}
lowercaseOutputName: true
lowercaseOutputLabelNames: false
whitelistObjectNames:
  - "java.lang:*"
  - "org.apache.flink.metrics:*"
rules:
  - pattern: "^java.lang.*"
# Kafka source

ogibayashi

2016-10-06 17:10:27 +0900 [warn]: temporarily failed to flush the buffer. next_retry=2016-10-06 17:10:28 +0900 error_class="NoMethodError" error="undefined method `[]=' for nil:NilClass" plugin_id="to_kafka"
  2016-10-06 17:10:27 +0900 [warn]: /home/edge-dev/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluent-plugin-kafka-0.2.2/lib/fluent/plugin/out_kafka_buffered.rb:215:in `block in write'
  2016-10-06 17:10:27 +0900 [warn]: /home/edge-dev/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluentd-0.12.28/lib/fluent/buffer.rb:123:in `each'
  2016-10-06 17:10:27 +0900 [warn]: /home/edge-dev/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluentd-0.12.28/lib/fluent/buffer.rb:123:in `block in msgpack_each'
  2016-10-06 17:10:27 +0900 [warn]: /home/edge-dev/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluentd-0.12.28/lib/fluent/plugin/buf_file.rb:71:in `open'
  2016-10-06 17:10:27 +0900 [warn]: /home/edge-dev/local/ruby-2.2/lib/ruby/gems/2.2.0/gems/fluentd-0.12.28/lib/fluent/buffer.rb:120:in `msgpack_each'
  2016-10-06 17:10:27 +0900 [w

ogibayashi

@@ -114,6 +116,7 @@ public class FlinkKafkaConsumer09<T> extends FlinkKafkaConsumerBase<T> {
 	/** If the consumer doesn't have a Kafka partition assigned at runtime, it'll block on this waitThread **/
 	private transient Thread waitThread;
 
+        private final ReentrantLock lock = new ReentrantLock(true);
 
 	// ------------------------------------------------------------------------
 
@@ -388,9 +391,12 @@ public class FlinkKafkaConsumer09<T> extends FlinkKafkaConsumerBase<T> {

ogibayashi

import java.sql.Timestamp
import java.util.concurrent.TimeUnit
import org.apache.flink.streaming.api.functions.source.FileMonitoringFunction
import org.apache.flink.streaming.api.functions.source.FileMonitoringFunction.{ WatchType}

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger

object DistinctCountTest {

ogibayashi

#!/usr/bin/env ruby

# Send process cpu usage for each fluentd process to fluentd.
# Each fluentd must be exected with -c (config) option
# This script might be executed periodically by cron.

require 'set'
require 'optparse'
require 'fluent-logger'

ogibayashi

gem 'norikra-udf-lookup'

ogibayashi

HDP2.2を手元のVMで試しにセットアップしてみたが、色々ハマった部分があったのでメモ

環境

CentOS6.3のVMを7つ用意して、以下のようにHA含めて構成することにした.

• master1: NameNode(active), ZKFC, JournalNode, Zookeeper
• master2: NameNode(standby), ZKFC, JournalNode, ResourceManager(standby), Zookeeper
• master3: JournalNode, ResourceManager(active), Zookeeper, HiveServer2, MySQL
• slaves(3ノード): DataNode, NodeManager

ogibayashi

Fluentdベンチマークのメモ

単一プロセスのin_tail

以下のように、ファイルを読み込みflowcountだけ取ってnullに捨てる

<source>
 type tail
 path /tmp/dummy_log.log

ogibayashi

--- iterator.rb.bak     2014-10-15 11:09:30.378246696 +0900
+++ iterator.rb 2014-10-15 11:48:56.200251049 +0900
@@ -29,6 +29,7 @@
     def each(&block)
       return self unless block_given?
       if current = self.next
+        break if range? && current[0] >= @_range[1]
         block[*current]
       end while valid?
       @_range = nil

ogibayashi

この本について

OpenShiftの管理者ではなく、Webアプリケーション開発者向けの本. OpenShift Onlineを使って、どのようにWebアプリケーションを動かすことができるか、ということが書いてある

1. Introduction

• OpenShiftとは？
  • RedHatが提供するPaaS
• 3つのバージョンがある
  • OpenShift Origin : オープンソースであり、最新版. 自分の環境に入れて使うことができる. OnlineやEnterpriseのUpstreamとなる.
  • OpenShift Online : RedHatが提供するクラウドサービス版. AWS上で動いており、アカウントを作ればOpenShiftの環境を使うことができる. 本書はこれを対象に書かれている.