# from __future__ import print_function
# import datetime
# import time
# import pymysql
# from pyspark.sql import SparkSession
# from pyspark.sql import SQLContext
# from pyspark import SparkConf,SparkContext
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from pyspark.context import SparkContext
from pyspark.conf import SparkConf
from pyspark.streaming import StreamingContext
from pyspark.sql import SQLContext
from pyspark.streaming.kafka import KafkaUtils
import argparse
import time
from datetime import datetime
import msgpack
from tensorflowonspark import TFCluster
import json
import pymysql
import tensorflow as tf




# def fetch_data(start_date, end_date):
#     # sc = SparkSession.builder.appName("Python Spark SQL basic example") \
#     #     .config('spark.some.config,option0', 'some-value') \
#     #     .getOrCreate()
#     sc = SparkContext(conf=SparkConf().setAppName("mnist_streaming"))
#     ctx = SQLContext(sc)
#     # jdbcDf = ctx.read.format("jdbc").options(url="jdbc:mysql://192.168.15.12:4000",
#     #                                          driver="com.mysql.jdbc.Driver",
#     #                                          # dbtable="((select device_id,cid_id,time,device_type,city_id,1 as clicked from jerry_test.data_feed_click where cid_id in (select id from eagle.src_mimas_prod_api_diary where doctor_id is not null and content_level >3.5)  and  cid_type = 'diary' and stat_date = '2018-08-12') union (select device_id,cid_id,time,device_type,city_id,0 as clicked from jerry_test.data_feed_exposure where cid_id in (select id from eagle.src_mimas_prod_api_diary where doctor_id is not null and content_level >3.5) and  cid_type = 'diary' and stat_date = '2018-08-12')) tmp",user="root",
#     #                                          dbtable="(select id as diary_id,doctor_id from eagle.src_mimas_prod_api_diary where doctor_id is not null and content_level >3.5 and datediff(current_date,created_time)<90) tmp",
#     #                                          user="root",
#     #                                          password="").load()
#     # df = ctx.read.format("jdbc").options(url="jdbc:mysql://rdsmaqevmuzj6jy.mysql.rds.aliyuncs.com:3306/doris_test",
#     #                                      driver="com.mysql.jdbc.Driver",
#     #                                      dbtable="device diary_queue",
#     #                                      user="work", password="workwork").load()
#     # df = ctx.read.format("jdbc").options(url="jdbc:mysql://rm-m5e842126ng59jrv6.mysql.rds.aliyuncs.com:3306/doris_prod",
#     #                                      driver="com.mysql.jdbc.Driver",
#     #                                      dbtable="device diary_queue",
#     #                                      user="doris", password="o5gbA27hXHHm").load()
#
#     jdbcDf = ctx.read.format("jdbc").options(url="jdbc:mysql://192.168.15.12:4000",
#                                              driver="com.mysql.jdbc.Driver",
#                                              dbtable = "(select device_id from data_feed_click limit 8) tmp",
#                                              user="root",password = "3SYz54LS9#^9sBvC").load()
#     jdbcDf.show(6)
#
#     # url = "jdbc:mysql://10.66.157.22:4000/jerry_prod"
#     # table = "data_feed_click"
#     # properties = {"user": "root", "password": "3SYz54LS9#^9sBvC"}
#     # df = sqlContext.read.jdbc(url, table, properties)




# def hello(args):
#     import tensorflow as tf
#     from tensorflow.contrib.boosted_trees.estimator_batch.estimator import GradientBoostedDecisionTreeClassifier
#     from tensorflow.contrib.boosted_trees.proto import learner_pb2 as gbdt_learner
#
#     # Ignore all GPUs (current TF GBDT does not support GPU).
#     import os
#     os.environ["CUDA_VISIBLE_DEVICES"] = ""
#
#     # Import MNIST data
#     # Set verbosity to display errors only (Remove this line for showing warnings)
#     tf.logging.set_verbosity(tf.logging.ERROR)
#     from tensorflow.examples.tutorials.mnist import input_data
#     mnist = input_data.read_data_sets("/tmp/data/", one_hot=False,
#                                       source_url='http://yann.lecun.com/exdb/mnist/')
#
#     # Parameters
#     batch_size = 10000# The number of samples per batch
#     num_classes = 10  # The 10 digits
#     num_features = 784  # Each image is 28x28 pixels
#     max_steps = 10000
#
#     # GBDT Parameters
#     learning_rate = 0.1
#     l1_regul = 0.
#     l2_regul = 1.
#     examples_per_layer = 1000
#     num_trees = 10
#     max_depth = 16
#
#     # Fill GBDT parameters into the config proto
#     learner_config = gbdt_learner.LearnerConfig()
#     learner_config.learning_rate_tuner.fixed.learning_rate = learning_rate
#     learner_config.regularization.l1 = l1_regul
#     learner_config.regularization.l2 = l2_regul / examples_per_layer
#     learner_config.constraints.max_tree_depth = max_depth
#     growing_mode = gbdt_learner.LearnerConfig.LAYER_BY_LAYER
#     learner_config.growing_mode = growing_mode
#     run_config = tf.contrib.learn.RunConfig(save_checkpoints_secs=300)
#     learner_config.multi_class_strategy = (
#         gbdt_learner.LearnerConfig.DIAGONAL_HESSIAN) \
#  \
#         # Create a TensorFlor GBDT Estimator
#     gbdt_model = GradientBoostedDecisionTreeClassifier(
#         model_dir=None,  # No save directory specified
#         learner_config=learner_config,
#         n_classes=num_classes,
#         examples_per_layer=examples_per_layer,
#         num_trees=num_trees,
#         center_bias=False,
#         config=run_config)
#
#     # Display TF info logs
#     tf.logging.set_verbosity(tf.logging.INFO)
#
#     # Define the input function for training
#     input_fn = tf.estimator.inputs.numpy_input_fn(
#         x={'images': mnist.train.images}, y=mnist.train.labels,
#         batch_size=batch_size, num_epochs=None, shuffle=True)
#     # Train the Model
#     gbdt_model.fit(input_fn=input_fn, max_steps=max_steps)
#
#     # Evaluate the Model
#     # Define the input function for evaluating
#     input_fn = tf.estimator.inputs.numpy_input_fn(
#         x={'images': mnist.test.images}, y=mnist.test.labels,
#         batch_size=batch_size, shuffle=False)
#     # Use the Estimator 'evaluate' method
#     e = gbdt_model.evaluate(input_fn=input_fn)
#
#     print("Testing Accuracy:", e['accuracy'])


if __name__ == "__main__":
    sc = SparkContext(
        conf=SparkConf().setAppName("mnist_streaming").set("spark.streaming.kafka.maxRatePerPartition", 100))
    ctx = SQLContext(sc)
    jdbcDf = ctx.read.format("jdbc").options(url="jdbc:mysql://rm-m5e842126ng59jrv6.mysql.rds.aliyuncs.com:3306/doris_prod",
                                             driver="com.mysql.jdbc.Driver",
                                             # dbtable="((select device_id,cid_id,time,device_type,city_id,1 as clicked from jerry_test.data_feed_click where cid_id in (select id from eagle.src_mimas_prod_api_diary where doctor_id is not null and content_level >3.5)  and  cid_type = 'diary' and stat_date = '2018-08-12') union (select device_id,cid_id,time,device_type,city_id,0 as clicked from jerry_test.data_feed_exposure where cid_id in (select id from eagle.src_mimas_prod_api_diary where doctor_id is not null and content_level >3.5) and  cid_type = 'diary' and stat_date = '2018-08-12')) tmp",user="root",
                                             dbtable="(select device_id from device_diary_queue limit 8) tmp",
                                             user="doris",
                                             password="o5gbA27hXHHm"
                                             ).load()
    jdbcDf.show(6)
    # fetch_data("2018-11-11","2018-11-12")
  # from pyspark.context import SparkContext
  # from pyspark.conf import SparkConf
  # from tensorflowonspark import TFCluster
  # import argparse
  #
  # sc = SparkContext(conf=SparkConf().setAppName("mnist_spark"))
  # executors = sc._conf.get("spark.executor.instances")
  # num_executors = int(executors) if executors is not None else 1
  #
  # parser = argparse.ArgumentParser()
  # parser.add_argument("--batch_size", help="number of records per batch", type=int, default=100)
  # parser.add_argument("--cluster_size", help="number of nodes in the cluster", type=int, default=num_executors)
  # parser.add_argument("--data_dir", help="path to MNIST data", default="MNIST-data")
  # parser.add_argument("--model", help="path to save model/checkpoint", default="mnist_model")
  # parser.add_argument("--num_ps", help="number of PS nodes in cluster", type=int, default=1)
  # parser.add_argument("--steps", help="maximum number of steps", type=int, default=1000)
  # parser.add_argument("--tensorboard", help="launch tensorboard process", action="store_true")
  #
  # args = parser.parse_args()
  # print("args:", args)
  #
  # cluster = TFCluster.run(sc, hello, args, args.cluster_size, args.num_ps, tensorboard=args.tensorboard, input_mode=TFCluster.InputMode.TENSORFLOW, log_dir=args.model, master_node='master')
  # cluster.shutdown()


