test.py

# 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





# 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__":
    from pyspark.sql import SQLContext
    from pyspark.context import SparkContext
    from pyspark.conf import SparkConf

    sc = SparkContext(conf=SparkConf().setAppName("mnist_streaming")).getOrCreate()
    ctx = SQLContext(sc)
    jdbcDf = ctx.read.format("jdbc").options(url="jdbc:mysql://192.168.15.12:4000/jerry_test",
                                             driver="com.mysql.jdbc.Driver",
                                             dbtable="(select * from nd_cid_similarity_matrix) tmp",
                                             user="root",
                                             password="").load()

    print(jdbcDf.printSchema())

    print(jdbcDf.collect())
    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()