exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleProducer "$@"

val producer = new KafkaProducer[Array[Byte], Array[Byte]](producerProps(config))

do {
  record = reader.readMessage()
  if (record != null)
    send(producer, record, config.sync)
} while (record != null)

private def send(producer: KafkaProducer[Array[Byte], Array[Byte]],
                       record: ProducerRecord[Array[Byte], Array[Byte]], sync: Boolean): Unit = {
  if (sync)
    producer.send(record).get()
  else
    producer.send(record, new ErrorLoggingCallback(record.topic, record.key, record.value, false))
}

String clientId = config.getString(ProducerConfig.CLIENT_ID_CONFIG);
if (clientId.length() <= 0)
    clientId = "producer-" + PRODUCER_CLIENT_ID_SEQUENCE.getAndIncrement();
this.clientId = clientId;

String transactionalId = userProvidedConfigs.containsKey(ProducerConfig.TRANSACTIONAL_ID_CONFIG) ?
        (String) userProvidedConfigs.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG) : null;

this.accumulator = new RecordAccumulator(logContext,
        config.getInt(ProducerConfig.BATCH_SIZE_CONFIG),
        this.compressionType,
        lingerMs(config),
        retryBackoffMs,
        deliveryTimeoutMs,
        metrics,
        PRODUCER_METRIC_GROUP_NAME,
        time,
        apiVersions,
        transactionManager,
        new BufferPool(this.totalMemorySize, config.getInt(ProducerConfig.BATCH_SIZE_CONFIG), metrics, time, PRODUCER_METRIC_GROUP_NAME));

this.sender = newSender(logContext, kafkaClient, this.metadata);

return new Sender(logContext,
        client,
        metadata,
        this.accumulator,
        maxInflightRequests == 1,
        producerConfig.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG),
        acks,
        retries,
        metricsRegistry.senderMetrics,
        time,
        requestTimeoutMs,
        producerConfig.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG),
        this.transactionManager,
        apiVersions);

exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleConsumer "$@"

val conf = new ConsumerConfig(args)
try {
  run(conf)

val consumer = new KafkaConsumer(consumerProps(conf), new ByteArrayDeserializer, new ByteArrayDeserializer)

addShutdownHook(consumerWrapper, conf)

def addShutdownHook(consumer: ConsumerWrapper, conf: ConsumerConfig) {
  Runtime.getRuntime.addShutdownHook(new Thread() {
    override def run() {
      consumer.wakeup()

      shutdownLatch.await()

      if (conf.enableSystestEventsLogging) {
        System.out.println("shutdown_complete")
      }
    }
  })
}

try process(conf.maxMessages, conf.formatter, consumerWrapper, System.out, conf.skipMessageOnError)

def process(maxMessages: Integer, formatter: MessageFormatter, consumer: ConsumerWrapper, output: PrintStream,
            skipMessageOnError: Boolean) {

(snip)

    val msg: ConsumerRecord[Array[Byte], Array[Byte]] = try {
      consumer.receive()

(snip)

def receive(): ConsumerRecord[Array[Byte], Array[Byte]] = {
  if (!recordIter.hasNext) {
    recordIter = consumer.poll(Duration.ofMillis(timeoutMs)).iterator
    if (!recordIter.hasNext)
      throw new TimeoutException()
  }

  recordIter.next
}

formatter.writeTo(new ConsumerRecord(msg.topic, msg.partition, msg.offset, msg.timestamp,
                                     msg.timestampType, 0, 0, 0, msg.key, msg.value, msg.headers), output)

  public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener) {

(snip)

  public void subscribe(Pattern pattern, ConsumerRebalanceListener listener) {

(snip)

public void assign(Collection<TopicPartition> partitions) {

private enum SubscriptionType {
    NONE, AUTO_TOPICS, AUTO_PATTERN, USER_ASSIGNED
}

public void updatePatternSubscription(Cluster cluster) {
    final Set<String> topicsToSubscribe = cluster.topics().stream()
            .filter(subscriptions::matchesSubscribedPattern)
            .collect(Collectors.toSet());
    if (subscriptions.subscribeFromPattern(topicsToSubscribe))
        metadata.requestUpdateForNewTopics();
}

private Node coordinator = null;

AbstractCoordinator.this.coordinator = new Node(
        coordinatorConnectionId,
        findCoordinatorResponse.data().host(),
        findCoordinatorResponse.data().port());

doJoinGroup(group, memberId, groupInstanceId, clientId, clientHost, rebalanceTimeoutMs, sessionTimeoutMs, protocolType, protocols, responseCallback)

private[group] class MemberMetadata(var memberId: String,
                                    val groupId: String,
                                    val groupInstanceId: Option[String],
                                    val clientId: String,
                                    val clientHost: String,
                                    val rebalanceTimeoutMs: Int,
                                    val sessionTimeoutMs: Int,
                                    val protocolType: String,
                                    var supportedProtocols: List[(String, Array[Byte])]) {

$ /opt/Anaconda/default/bin/conda create -n mlflow ipython jupyter
$ source activate mlflow
$ pip install mlflow

$ mkdir -p Sources/mlflow_quickstart
$ cd Sources/mlflow_quickstart
$ cat << EOF > quickstart.py 
> import os
> from mlflow import log_metric, log_param, log_artifact
> 
> if __name__ == "__main__":
>     # Log a parameter (key-value pair)
>     log_param("param1", 5)
> 
>     # Log a metric; metrics can be updated throughout the run
>     log_metric("foo", 1)
>     log_metric("foo", 2)
>     log_metric("foo", 3)
> 
>     # Log an artifact (output file)
>     with open("output.txt", "w") as f:
>         f.write("Hello world! 1\n")
>     with open("output.txt", "w") as f:
>         f.write("Hello world! 2\n")
>     log_artifact("output.txt")
>     with open("output.txt", "w") as f:
>         f.write("Hello world! 3\n")
>     log_artifact("output.txt")
> EOF

$ ls
mlruns  output.txt  quickstart.py

$ mlflow ui

$ mlflow run https://github.com/mlflow/mlflow-example.git -P alpha=5

wine_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "wine-quality.csv")
data = pd.read_csv(wine_path)

with mlflow.start_run():

lr = ElasticNet(alpha=alpha, l1_ratio=l1_ratio, random_state=42)
lr.fit(train_x, train_y)

predicted_qualities = lr.predict(test_x)

(rmse, mae, r2) = eval_metrics(test_y, predicted_qualities)

mlflow.log_param("alpha", alpha)
mlflow.log_param("l1_ratio", l1_ratio)
mlflow.log_metric("rmse", rmse)
mlflow.log_metric("r2", r2)
mlflow.log_metric("mae", mae)

mlflow.sklearn.log_model(lr, "model")

for alpha in 0 1 0.5
do
  for l1_ratio in 1 0.5 0.2 0
  do
    python ~/Sources/mlflow/examples/sklearn_elasticnet_wine/train.py ${alpha} ${l1_ratio}
  done
done

name: tutorial

conda_env: conda.yaml

entry_points:
  main:
    parameters:
      alpha: float
      l1_ratio: {type: float, default: 0.1}
    command: "python train.py {alpha} {l1_ratio}"

name: tutorial
channels:
  - defaults
dependencies:
  - python=3.6
  - scikit-learn=0.19.1
  - pip:
    - mlflow>=1.0

$ mlflow run ~/Sources/mlflow/examples/sklearn_elasticnet_wine -P alpha=0.5

$ cd ~/Sources/mlflow/examples/docker
$ docker build -t mlflow-docker-example -f Dockerfile .

$ sudo mlflow run ~/Sources/mlflow/examples/docker -P alpha=0.5

name: docker-example

docker_env:
  image:  mlflow-docker-example

entry_points:
  main:
    parameters:
      alpha: float
      l1_ratio: {type: float, default: 0.1}
    command: "python train.py --alpha {alpha} --l1-ratio {l1_ratio}"

$ mlflow models serve -m /home/dobachi/Sources/mlflow_tutorials/mlruns/0/a87ee8c6c6f04f5c822a32e3ecae830e/artifacts/model -p 1234

$ curl -X POST -H "Content-Type:application/json; format=pandas-split" --data '{"columns":["fixed acidity","volatile acidity","citric acid","residual sugar","chlorides","free sulfur dioxide","total sulfur dioxide","density","pH","sulphates","alcohol"],"data":[[7,0.27,0.36,20.7,0.045,45,170,1.001,3,0.45,8.8]]}' http://127.0.0.1:1234/invocations

$ mlflow server --backend-store-uri /tmp/hoge --default-artifact-root /tmp/fuga --host 0.0.0.0

$ MLFLOW_TRACKING_URI=http://localhost:5000 mlflow run ~/Sources/mlflow/examples/sklearn_elasticnet_wine -P alpha=0.5

import re
import sys

from mlflow.cli import cli

if __name__ == '__main__':
    sys.argv[0] = re.sub(r'(-script\.pyw?|\.exe)?$', '', sys.argv[0])
    sys.exit(cli())

cli.add_command(mlflow.models.cli.commands)
cli.add_command(mlflow.sagemaker.cli.commands)
cli.add_command(mlflow.experiments.commands)
cli.add_command(mlflow.store.cli.commands)
cli.add_command(mlflow.azureml.cli.commands)
cli.add_command(mlflow.runs.commands)
cli.add_command(mlflow.db.commands)

with mlflow.start_run():
    lr = ElasticNet(alpha=alpha, l1_ratio=l1_ratio, random_state=42)
    lr.fit(train_x, train_y)

start_run = mlflow.tracking.fluent.start_run

_active_run_stack.append(ActiveRun(active_run_obj))
return _active_run_stack[-1]

class ActiveRun(Run):  # pylint: disable=W0223
    """Wrapper around :py:class:`mlflow.entities.Run` to enable using Python ``with`` syntax."""

    def __init__(self, run):
        Run.__init__(self, run.info, run.data)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        status = RunStatus.FINISHED if exc_type is None else RunStatus.FAILED
        end_run(RunStatus.to_string(status))
        return exc_type is None

def ui(backend_store_uri, default_artifact_root, port):
    """
    Launch the MLflow tracking UI for local viewing of run results. To launch a production
    server, use the "mlflow server" command instead.

    The UI will be visible at http://localhost:5000 by default.

_run_server(backend_store_uri, default_artifact_root, "127.0.0.1", port, 1, None, [])

def _run_server(file_store_path, default_artifact_root, host, port, workers, static_prefix,
                gunicorn_opts):
    """
    Run the MLflow server, wrapping it in gunicorn
    :param static_prefix: If set, the index.html asset will be served from the path static_prefix.
                          If left None, the index.html asset will be served from the root path.
    :return: None
    """
    env_map = {}
    if file_store_path:
        env_map[BACKEND_STORE_URI_ENV_VAR] = file_store_path
    if default_artifact_root:
        env_map[ARTIFACT_ROOT_ENV_VAR] = default_artifact_root
    if static_prefix:
        env_map[STATIC_PREFIX_ENV_VAR] = static_prefix
    bind_address = "%s:%s" % (host, port)
    opts = shlex.split(gunicorn_opts) if gunicorn_opts else []
    exec_cmd(["gunicorn"] + opts + ["-b", bind_address, "-w", "%s" % workers, "mlflow.server:app"],
             env=env_map, stream_output=True)

Usage: mlflow server [OPTIONS]

  Run the MLflow tracking server.

  The server which listen on http://localhost:5000 by default, and only
  accept connections from the local machine. To let the server accept
  connections from other machines, you will need to pass --host 0.0.0.0 to
  listen on all network interfaces (or a specific interface address).

Options:
  --backend-store-uri PATH     URI to which to persist experiment and run
                               data. Acceptable URIs are SQLAlchemy-compatible
                               database connection strings (e.g.
                               'sqlite:///path/to/file.db') or local
                               filesystem URIs (e.g.
                               'file:///absolute/path/to/directory'). By
                               default, data will be logged to the ./mlruns
                               directory.
  --default-artifact-root URI  Local or S3 URI to store artifacts, for new
                               experiments. Note that this flag does not
                               impact already-created experiments. Default:
                               Within file store, if a file:/ URI is provided.
                               If a sql backend is used, then this option is
                               required.
  -h, --host HOST              The network address to listen on (default:
                               127.0.0.1). Use 0.0.0.0 to bind to all
                               addresses if you want to access the tracking
                               server from other machines.
  -p, --port INTEGER           The port to listen on (default: 5000).
  -w, --workers INTEGER        Number of gunicorn worker processes to handle
                               requests (default: 4).
  --static-prefix TEXT         A prefix which will be prepended to the path of
                               all static paths.
  --gunicorn-opts TEXT         Additional command line options forwarded to
                               gunicorn processes.
  --help                       Show this message and exit.

exec_cmd(["gunicorn"] + opts + ["-b", bind_address, "-w", "%s" % workers, "mlflow.server:app"],
         env=env_map, stream_output=True)

# Serve the index.html for the React App for all other routes.
@app.route(_add_static_prefix('/'))
def serve():
    return send_from_directory(STATIC_DIR, 'index.html')

REL_STATIC_DIR = "js/build"

app = Flask(__name__, static_folder=REL_STATIC_DIR)
STATIC_DIR = os.path.join(app.root_path, REL_STATIC_DIR)

name: tutorial

conda_env: conda.yaml

entry_points:
  main:
    parameters:
      alpha: float
      l1_ratio: {type: float, default: 0.1}
    command: "python train.py {alpha} {l1_ratio}"

alpha = float(sys.argv[1]) if len(sys.argv) > 1 else 0.5
l1_ratio = float(sys.argv[2]) if len(sys.argv) > 2 else 0.5

name: tutorial
channels:
  - defaults
dependencies:
  - numpy=1.14.3
  - pandas=0.22.0
  - scikit-learn=0.19.1
  - pip:
    - mlflow

# Read the wine-quality csv file (make sure you're running this from the root of MLflow!)
wine_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "wine-quality.csv")
data = pd.read_csv(wine_path)

# Split the data into training and test sets. (0.75, 0.25) split.
train, test = train_test_split(data)

# The predicted column is "quality" which is a scalar from [3, 9]
train_x = train.drop(["quality"], axis=1)
test_x = test.drop(["quality"], axis=1)
train_y = train[["quality"]]
test_y = test[["quality"]]

with mlflow.start_run():

lr = ElasticNet(alpha=alpha, l1_ratio=l1_ratio, random_state=42)
lr.fit(train_x, train_y)

predicted_qualities = lr.predict(test_x)

(rmse, mae, r2) = eval_metrics(test_y, predicted_qualities)

Elasticnet model (alpha=5.000000, l1_ratio=0.100000):
  RMSE: 0.8594260117338262
  MAE: 0.6480675144220314
  R2: 0.046025292604596424

mlflow.log_param("alpha", alpha)
mlflow.log_param("l1_ratio", l1_ratio)
mlflow.log_metric("rmse", rmse)
mlflow.log_metric("r2", r2)
mlflow.log_metric("mae", mae)

mlflow.sklearn.log_model(lr, "model")

$ cd Sources
$ git clone https://github.com/mlflow/mlflow.git
$ cd mlflow/examples/docker

diff --git a/examples/docker/Dockerfile b/examples/docker/Dockerfileindex e436f49..686e0e2 100644--- a/examples/docker/Dockerfile+++ b/examples/docker/Dockerfile@@ -1,5 +1,7 @@ FROM continuumio/miniconda:4.5.4+RUN conda install ipython jupyter
+ RUN pip install mlflow>=1.0 \
     && pip install azure-storage==0.36.0 \
     && pip install numpy==1.14.3

$ sudo -i docker build -t "dobachi/mlflow:latest" `pwd`

$ cat << EOF > tutorial.py
> import os
> from mlflow import log_metric, log_param, log_artifact
>
> if __name__ == "__main__":
>     # Log a parameter (key-value pair)
>     log_param("param1", 5)
>
>     # Log a metric; metrics can be updated throughout the run
>     log_metric("foo", 1)
>     log_metric("foo", 2)
>     log_metric("foo", 3)
>
>     # Log an artifact (output file)
>     with open("output.txt", "w") as f:
>         f.write("Hello world!")
>     log_artifact("output.txt")
> EOF

$ sudo -i docker run -v `pwd`:/apps --rm -it dobachi/mlflow:latest  /bin/bash

# cd apps
# python tutorial.py

# cat output.txt  
Hello world!