$ confluent-hub install jcustenborder/kafka-connect-twitter

name=connector1
tasks.max=1
connector.class=com.github.jcustenborder.kafka.connect.twitter.TwitterSourceConnector

# Set these required values
twitter.oauth.accessTokenSecret=hoge
process.deletes=false
filter.keywords=kafka
kafka.status.topic=twitter-status
kafka.delete.topic=twitter-delete
twitter.oauth.consumerSecret=hoge
twitter.oauth.accessToken=hoge
twitter.oauth.consumerKey=hoge

$ connect-standalone /etc/kafka/connect-standalone.properties /etc/kafka/connect-twitter-source.properties

curl -H "Content-Type: application/json" -X POST http://localhost:8083/connectors -d '
{
    "name": "twitter",
    "config": {
        "tasks.max":1,
        "connector.class":"com.github.jcustenborder.kafka.connect.twitter.TwitterSourceConnector",
        "twitter.oauth.accessTokenSecret":"hoge",
        "process.deletes":"false",
        "filter.keywords":"kafka",
        "kafka.status.topic":"twitter-status",
        "kafka.delete.topic":"twitter-delete",
        "twitter.oauth.consumerSecret":"hoge",
        "twitter.oauth.accessToken":"hoge",
        "twitter.oauth.consumerKey":"hoge",
    }
}
'

$ kafka-console-consumer --bootstrap-server broker:9092 --topic twitter-status | jq .

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "int64",
        "optional": true,
        "name": "org.apache.kafka.connect.data.Timestamp",
        "version": 1,
        "doc": "Return the created_at",
        "field": "CreatedAt"
      },

(snip)

  "payload": {
    "CreatedAt": XXXXXXXXXXXXX,
    "Id": XXXXXXXXXXXXXXXXXXX,
    "Text": "hoge",
    "Source": "<a href=\"http://twitter.com/download/android\" rel=\"nofollow\">Twitter for Android</a>",
    "Truncated": false,

(snip)

$ ls /var/lib/kafka/data/twitter-status-0
00000000000000000000.index  00000000000000000000.log  00000000000000000000.timeindex  leader-epoch-checkpoint

$ kafka-run-class kafka.tools.DumpLogSegments --deep-iteration --print-data-log --files /var/lib/kafka/data/twitter-status-0/00000000000000000000.index | head

Dumping /var/lib/kafka/data/twitter-status-0/00000000000000000000.index
offset: 1 position: 16674
offset: 2 position: 33398
offset: 3 position: 50562
offset: 4 position: 67801

(snip)

$ kafka-run-class kafka.tools.DumpLogSegments --deep-iteration --print-data-log --files /var/lib/kafka/data/twitter-status-0/00000000000000000000.log | head -n 4

Dumping /var/lib/kafka/data/twitter-status-0/00000000000000000000.log
Starting offset: 0
offset: 0 position: 0 CreateTime: 1552055018454 isvalid: true keysize: 239 valuesize: 16362 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: {"schema":{"type":"struct","fields":[{"type":"int64","optional":true,"field":"Id"}],"optional":false,"name":"com.github.jcustenborder.kafka.connect.twitter.StatusKey","doc":"Key for a twitter status."

(snip)

offset: 1 position: 16674 CreateTime: 1552055018465 isvalid: true keysize: 239 valuesize: 16412 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: {"schema":{"type":"struct","fields":[{"type":"int64","optional":true,"field":"Id"}],"optional":false,"name":"com.github.jcustenborder.kafka.connect.twitter.StatusKey","doc":"Key for a twitter status."},"payload":{"Id":1104024860143968257}

(snip)

$ kafka-run-class kafka.tools.DumpLogSegments --deep-iteration --print-data-log --files /var/lib/kafka/data/twitter-status-0/00000000000000000000.timeindex | head -n 10

Dumping /var/lib/kafka/data/twitter-status-0/00000000000000000000.timeindextimestamp: 1552055018465 offset: 1timestamp: 1552055529166 offset: 2timestamp: 1552055536284 offset: 3timestamp: 1552055626862 offset: 4timestamp: 1552055652086 offset: 5timestamp: 1552055717443 offset: 6timestamp: 1552055788403 offset: 7timestamp: 1552055789505 offset: 8

(snip)

bash-4.4# zookeeper-shell.sh zookeeper:2181
Connecting to zookeeper:2181

ls /kafka
[log_dir_event_notification, isr_change_notification, admin, consumers, cluster, config, latest_producer_id_block, controller, brokers, controller_epoch]

bash-4.4#  kafka-topics.sh --create --topic topic --partitions 1 --zookeeper zookeeper:2181/kafka --replication-factor 1

bash-4.4# kafka-topics.sh --topic topic --zookeeper zookeeper:2181/kafka --describeTopic:topic     PartitionCount:1        ReplicationFactor:1     Configs:
        Topic: topic    Partition: 0    Leader: 1001    Replicas: 1001  Isr: 1001

bash-4.4# kafka-console-producer.sh --topic=topic --broker-list=kafka:9092
>hoge
>fuga
>hoge
>fuga

bash-4.4# kafka-console-consumer.sh --bootstrap-server kafka:9092 --from-beginning --topic topic
hoge
fuga
hoge
fuga

private class LogDirFailureHandler(name: String, haltBrokerOnDirFailure: Boolean) extends ShutdownableThread(name) {
  Override def doWork() {
    val newOfflineLogDir = logDirFailureChannel.takeNextOfflineLogDir()
    if (haltBrokerOnDirFailure) {
      fatal(s"Halting broker because dir $newOfflineLogDir is offline")
      Exit.halt(1)
    }
    handleLogDirFailure(newOfflineLogDir)
  }
}

def maybePropagateIsrChanges() {
  val now = System.currentTimeMillis()
  isrChangeSet synchronized {
    if (isrChangeSet.nonEmpty &&
      (lastIsrChangeMs.get() + ReplicaManager.IsrChangePropagationBlackOut < now ||
        lastIsrPropagationMs.get() + ReplicaManager.IsrChangePropagationInterval < now)) {
      zkClient.propagateIsrChanges(isrChangeSet)
      isrChangeSet.clear()
      lastIsrPropagationMs.set(now)
    }
  }
}

ls /kafka/brokers
[seqid, topics, ids]

def registerBroker(brokerInfo: BrokerInfo): Long = {
  val path = brokerInfo.path
  val stat = checkedEphemeralCreate(path, brokerInfo.toJsonBytes)
  info(s"Registered broker ${brokerInfo.broker.id} at path $path with addresses: ${brokerInfo.broker.endPoints}, czxid (broker epoch): ${stat.getCzxid}")
  stat.getCzxid
}

get /kafka/brokers/topics/topic/partitions/0/state
{"controller_epoch":1,"leader":1001,"version":1,"leader_epoch":0,"isr":[1001]}

get /kafka/controller
{"version":1,"brokerid":1001,"timestamp":"1551794212551"}

def validateTopicCreate(topic: String,
                        partitionReplicaAssignment: Map[Int, Seq[Int]],
                        config: Properties): Unit = {

createTopicPolicy match {
  case Some(policy) =>
    adminZkClient.validateTopicCreate(topic.name(), assignments, configs)

def getAllBrokerAndEpochsInCluster: Map[Broker, Long] = {
  val brokerIds = getSortedBrokerList
  val getDataRequests = brokerIds.map(brokerId => GetDataRequest(BrokerIdZNode.path(brokerId), ctx = Some(brokerId)))
  val getDataResponses = retryRequestsUntilConnected(getDataRequests)
  getDataResponses.flatMap { getDataResponse =>
    val brokerId = getDataResponse.ctx.get.asInstanceOf[Int]
    getDataResponse.resultCode match {
      case Code.OK =>
        Some((BrokerIdZNode.decode(brokerId, getDataResponse.data).broker, getDataResponse.stat.getCzxid))
      case Code.NONODE => None
      case _ => throw getDataResponse.resultException.get
    }
  }.toMap
}

if redis_ip is None:
    self.external_redis = False

clipper_conn = ClipperConnection(DockerContainerManager())
clipper_conn.start_clipper()
python_deployer.create_endpoint(clipper_conn, "simple-example", "doubles",
                                feature_sum)

def feature_sum(xs):
    return [str(sum(x)) for x in xs]

while True:
    if batch_size > 1:
        predict(
            clipper_conn.get_query_addr(),
            [list(np.random.random(200)) for i in range(batch_size)],
            batch=True)
    else:
        predict(clipper_conn.get_query_addr(), np.random.random(200))
    time.sleep(0.2)

def predict(addr, x, batch=False):
    url = "http://%s/simple-example/predict" % addr

    if batch:
        req_json = json.dumps({'input_batch': x})
    else:
        req_json = json.dumps({'input': list(x)})

    headers = {'Content-type': 'application/json'}
    start = datetime.now()
    r = requests.post(url, headers=headers, data=req_json)
    end = datetime.now()
    latency = (end - start).total_seconds() * 1000.0
    print("'%s', %f ms" % (r.text, latency))

(clipper) $ pip install xgboost
(clipper) $ ipython

import logging, xgboost as xgb, numpy as np
from clipper_admin import ClipperConnection, DockerContainerManager
cl = ClipperConnection(DockerContainerManager())
cl.start_clipper()

19-02-27:22:49:09 INFO     [docker_container_manager.py:119] Starting managed Redis instance in Docker
19-02-27:22:49:14 INFO     [clipper_admin.py:126] Clipper is running

cl.register_application('xgboost-test', 'integers', 'default_pred', 100000)

19-02-27:22:49:55 INFO     [clipper_admin.py:201] Application xgboost-test was successfully registered

def get_test_point():
    return [np.random.randint(255) for _ in range(784)]

# Create a training matrix.
dtrain = xgb.DMatrix(get_test_point(), label=[0])

# We then create parameters, watchlist, and specify the number of rounds
# This is code that we use to build our XGBoost Model, and your code may differ.
param = {'max_depth': 2, 'eta': 1, 'silent': 1, 'objective': 'binary:logistic'}
watchlist = [(dtrain, 'train')]
num_round = 2
bst = xgb.train(param, dtrain, num_round, watchlist)

[0]     train-error:0
[1]     train-error:0

def predict(xs):
    return bst.predict(xgb.DMatrix(xs))

from clipper_admin.deployers import python as python_deployer
# We specify which packages to install in the pkgs_to_install arg.
# For example, if we wanted to install xgboost and psycopg2, we would use
# pkgs_to_install = ['xgboost', 'psycopg2']
python_deployer.deploy_python_closure(cl, name='xgboost-model', version=1,
     input_type="integers", func=predict, pkgs_to_install=['xgboost'])

19-02-27:22:54:35 INFO     [deployer_utils.py:44] Saving function to /tmp/clipper/tmpincj4sg2
19-02-27:22:54:35 INFO     [deployer_utils.py:54] Serialized and supplied predict function
19-02-27:22:54:35 INFO     [python.py:192] Python closure saved

(snip)

19-02-27:22:54:53 INFO     [docker_container_manager.py:257] Found 0 replicas for xgboost-model:1. Adding 1
19-02-27:22:55:00 INFO     [clipper_admin.py:635] Successfully registered model xgboost-model:1
19-02-27:22:55:00 INFO     [clipper_admin.py:553] Done deploying model xgboost-model:1.

cl.link_model_to_app('xgboost-test', 'xgboost-model')

import requests, json
# Get Address
addr = cl.get_query_addr()
# Post Query
response = requests.post(
     "http://%s/%s/predict" % (addr, 'xgboost-test'),
     headers={"Content-type": "application/json"},
     data=json.dumps({
         'input': get_test_point()
     }))
result = response.json()
if response.status_code == requests.codes.ok and result["default"]:
     print('A default prediction was returned.')
elif response.status_code != requests.codes.ok:
    print(result)
    raise BenchmarkException(response.text)
else:
    print('Prediction Returned:', result)

Prediction Returned: {'query_id': 2, 'output': 0.3266071, 'default': False}

C++ 56.0%	 Python 24.1%	 CMake 8.1%	 Scala 3.0%	 Shell 2.5%	 Java 2.2%	 Other 4.1%

self.cm.start_clipper(query_frontend_image, mgmt_frontend_image,
                      frontend_exporter_image, cache_size,
                      num_frontend_replicas)

class ContainerManager(object):
    __metaclass__ = abc.ABCMeta

    @abc.abstractmethod
    def start_clipper(self, query_frontend_image, mgmt_frontend_image,

self.docker_client.networks.create(
    self.docker_network, check_duplicate=True)

if not self.external_redis:
    self.logger.info("Starting managed Redis instance in Docker")
    self.redis_port = find_unbound_port(self.redis_port)
    redis_labels = self.common_labels.copy()
    redis_labels[CLIPPER_DOCKER_PORT_LABELS['redis']] = str(
        self.redis_port)
    redis_container = self.docker_client.containers.run(
        'redis:alpine',
        "redis-server --port %s" % CLIPPER_INTERNAL_REDIS_PORT,
        name="redis-{}".format(random.randint(
            0, 100000)),  # generate a random name
        ports={
            '%s/tcp' % CLIPPER_INTERNAL_REDIS_PORT: self.redis_port
        },
        labels=redis_labels,
        **self.extra_container_kwargs)
    self.redis_ip = redis_container.name

mgmt_cmd = "--redis_ip={redis_ip} --redis_port={redis_port}".format(
    redis_ip=self.redis_ip, redis_port=CLIPPER_INTERNAL_REDIS_PORT)
self.clipper_management_port = find_unbound_port(
    self.clipper_management_port)
mgmt_labels = self.common_labels.copy()
mgmt_labels[CLIPPER_MGMT_FRONTEND_CONTAINER_LABEL] = ""
mgmt_labels[CLIPPER_DOCKER_PORT_LABELS['management']] = str(
    self.clipper_management_port)
self.docker_client.containers.run(
    mgmt_frontend_image,
    mgmt_cmd,
    name="mgmt_frontend-{}".format(random.randint(
        0, 100000)),  # generate a random name
    ports={
        '%s/tcp' % CLIPPER_INTERNAL_MANAGEMENT_PORT:
        self.clipper_management_port
    },
    labels=mgmt_labels,
    **self.extra_container_kwargs)

self.docker_client.containers.run(
    query_frontend_image,
    query_cmd,
    name=query_name,
    ports={
        '%s/tcp' % CLIPPER_INTERNAL_QUERY_PORT:
        self.clipper_query_port,
        '%s/tcp' % CLIPPER_INTERNAL_RPC_PORT: self.clipper_rpc_port
    },
    labels=query_labels,
    **self.extra_container_kwargs)

run_metric_image(self.docker_client, metric_labels,
                 self.prometheus_port, self.prom_config_path,
                 self.extra_container_kwargs)

def create_endpoint(clipper_conn,
                    name,
                    input_type,
                    func,
                    default_output="None",
                    version=1,
                    slo_micros=3000000,
                    labels=None,
                    registry=None,
                    base_image="default",
                    num_replicas=1,
                    batch_size=-1,
                    pkgs_to_install=None):

clipper_conn.register_application(name, input_type, default_output,
                                  slo_micros)
deploy_python_closure(clipper_conn, name, version, input_type, func,
                      base_image, labels, registry, num_replicas,
                      batch_size, pkgs_to_install)

clipper_conn.link_model_to_app(name, name)

url = "http://{host}/admin/add_app".format(
    host=self.cm.get_admin_addr())

const std::string ADD_APPLICATION = ADMIN_PATH + "/add_app$";

server_.add_endpoint(
    ADD_APPLICATION, "POST",
    [this](std::shared_ptr<HttpServer::Response> response,
           std::shared_ptr<HttpServer::Request> request) {
      try {
        clipper::log_info(LOGGING_TAG_MANAGEMENT_FRONTEND,
                          "Add application POST request");

def deploy_python_closure(clipper_conn,
                          name,
                          version,
                          input_type,
                          func,
                          base_image="default",
                          labels=None,
                          registry=None,
                          num_replicas=1,
                          batch_size=-1,
                          pkgs_to_install=None):

serialization_dir = save_python_function(name, func)

elif py_minor_version == (3, 6):
    logger.info("Using Python 3.6 base image")
    base_image = "{}/python36-closure-container:{}".format(
        __registry__, __version__)

clipper_conn.build_and_deploy_model(
    name, version, input_type, serialization_dir, base_image, labels,
    registry, num_replicas, batch_size, pkgs_to_install)

if not self.connected:
    raise UnconnectedException()
image = self.build_model(name, version, model_data_path, base_image,
                         container_registry, pkgs_to_install)
self.deploy_model(name, version, input_type, image, labels,
                  num_replicas, batch_size)

$ sudo yum install gcc
$ conda create -n clipper python=3.6 python
$ conda activate clipper
$ conda install ipython
$ pip install clipper_admin

(clipper) $ ipython

from clipper_admin import ClipperConnection, DockerContainerManager
clipper_conn = ClipperConnection(DockerContainerManager())
clipper_conn.start_clipper()

19-02-26:22:24:42 INFO     [docker_container_manager.py:119] Starting managed Redis instance in Docker
19-02-26:22:26:50 INFO     [clipper_admin.py:126] Clipper is running

clipper_conn.register_application(name="hello-world", input_type="doubles", default_output="-1.0", slo_micros=100000)

def feature_sum(xs):
    return [str(sum(x)) for x in xs]

from clipper_admin.deployers import python as python_deployer
python_deployer.deploy_python_closure(clipper_conn, name="sum-model", version=1, input_type="doubles", func=feature_sum)

19-02-26:22:30:59 INFO     [deployer_utils.py:44] Saving function to /tmp/clipper/tmp67eliqhx
19-02-26:22:30:59 INFO     [deployer_utils.py:54] Serialized and supplied predict function
19-02-26:22:30:59 INFO     [python.py:192] Python closure saved
19-02-26:22:30:59 INFO     [python.py:206] Using Python 3.6 base image

(snip)

19-02-26:22:31:26 INFO     [docker_container_manager.py:257] Found 0 replicas for sum-model:1. Adding 1
19-02-26:22:31:33 INFO     [clipper_admin.py:635] Successfully registered model sum-model:1
19-02-26:22:31:33 INFO     [clipper_admin.py:553] Done deploying model sum-model:1.
19-02-26:22:30:59 INFO     [clipper_admin.py:452] Building model Docker image with model data from /tmp/clipper/tmp67eliqhx

clipper_conn.link_model_to_app(app_name="hello-world", model_name="sum-model")

$ curl -X POST --header "Content-Type:application/json" -d '{"input": [1.1, 2.2, 3.3]}' 127.0.0.1:1337/hello-world/predict

{"query_id":0,"output":6.6,"default":false}

import requests, json, numpy as np
headers = {"Content-type": "application/json"}
requests.post("http://localhost:1337/hello-world/predict", headers=headers, data=json.dumps({"input": list(np.random.random(10))})).json()

Out[12]: {'query_id': 1, 'output': 4.710181343957851, 'default': False}

$ sudo docker ps
CONTAINER ID        IMAGE                               COMMAND                  CREATED             STATUS                    PORTS                                            NAMES
b041228b2a4d        sum-model:1                         "/container/containe…"   25 minutes ago      Up 25 minutes (healthy)                                                    sum-model_1-37382
6f93447357bd        prom/prometheus:v2.1.0              "/bin/prometheus --c…"   30 minutes ago      Up 30 minutes             0.0.0.0:9090->9090/tcp                           metric_frontend-37801
b3034a7352b8        clipper/frontend-exporter:0.3.0     "python /usr/src/app…"   30 minutes ago      Up 30 minutes                                                              query_frontend_exporter-33443
6b775a49e1ff        clipper/query_frontend:0.3.0        "/clipper/release/sr…"   30 minutes ago      Up 30 minutes             0.0.0.0:1337->1337/tcp, 0.0.0.0:7000->7000/tcp   query_frontend-33443
2c75406fda84        clipper/management_frontend:0.3.0   "/clipper/release/sr…"   30 minutes ago      Up 30 minutes             0.0.0.0:1338->1338/tcp                           mgmt_frontend-39690
ff14d91f313e        redis:alpine                        "docker-entrypoint.s…"   32 minutes ago      Up 32 minutes             0.0.0.0:6379->6379/tcp                           redis-28775

clipper_conn.stop_all()

Tellusとは
Tellus（テルース）」は、政府衛星データを利用した新たなビジネスマーケットプレイスを創出することを目的とした、日本初のオープン＆フリーな衛星データプラットフォームです。複数のデータをかけ合わせ、新たなビジネス創出を促進するためのあらゆるファンクションを提供します。

- name: check_state_of_wsl
  win_shell: Get-WindowsOptionalFeature -Online | ? FeatureName -Match "Microsoft-Windows-Subsystem-Linux" |  ConvertTo-Json
  register: wsl_check_json

- set_fact:
    wsl_check: "{{ wsl_check_json.stdout }}"
  
- name: enable_wsl
  win_shell: Enable-WindowsOptionalFeature -Online -FeatureName Microsoft-Windows-Subsystem-Linux -NoRestart
  when: wsl_check.State != 2

$ conda create -n ansible python
$ conda activate ansible
$ pip install ansible pywinrm

[win]
<対象となるWin環境>

[win:vars]
ansible_ssh_user=Administrator
ansible_ssh_port=5986
ansible_connection=winrm
ansible_winrm_transport=ntlm
ansible_winrm_server_cert_validation=ignore

$ ansible -i hosts.win -m win_ping win -k

$ ansible -i hosts.win -m win_chocolatey -a "name=googlechrome state=present" win -k
$ ansible -i hosts.win -m win_chocolatey -a "name=vim state=present" win -k

$ /opt/Anaconda/default/bin/conda create -n studioml python
$ conda activate stduioml
$ pip install studioml

awscli 1.16.101 has requirement rsa<=3.5.0,>=3.1.2, but you'll have rsa 4.0 which is incompatible.

$ cd ~/Sources
$ git clone https://github.com/studioml/studio.git
$ cd studio
$ cd examples/keras/

$ studio ui

  File "/home/******/.conda/envs/studioml/lib/python3.7/site-packages/flask/app.py", line 1813, in full_dispatch_request    rv = self.dispatch_request()  File "/home/******/.conda/envs/studioml/lib/python3.7/site-packages/flask/app.py", line 1799, in dispatch_request    return self.view_functions[rule.endpoint](**req.view_args)  File "/home/******/.conda/envs/studioml/lib/python3.7/site-packages/studio/apiserver.py", line 37, in dashboard    return _render('dashboard.html')  File "/home/******/.conda/envs/studioml/lib/python3.7/site-packages/studio/apiserver.py", line 518, in _render    auth = get_auth(get_auth_config())  File "/home/******/.conda/envs/studioml/lib/python3.7/site-packages/studio/apiserver.py", line 511, in get_auth_config    return get_config()['server']['authentication']

KeyError: 'server'

--- /home/dobachi/.studioml/config.yaml.2019021001      2019-02-10 01:04:10.751282800 +0900
+++ /home/dobachi/.studioml/config.yaml 2019-02-10 01:17:20.178761700 +0900
@@ -2,6 +2,10 @@
     type: http     serverUrl: https://zoo.studio.ml
     authentication: github
+    guest: true
+
+server:
+    authentication: None

 storage:
     type: gcloud

requests.exceptions.ConnectionError: HTTPSConnectionPool(host='zoo.studio.ml', port=443): Max retries exceeded with url: /api/get_user_experiments (Caused by NewConnectionError('<urllib3.connection.VerifiedHTTPSConnection object at 0x7ff5a64a3da0>: Failed to establisha new connection: [Errno -2] Name or service not known'))