Python中json.loads方法的典型用法代码

示例1: _initialize_members

# 需要导入模块: import json [as 别名]

# 或者: from json import loads [as 别名]

def _initialize_members(self, db_name, hdfs_app_path, kafka_consumer, conf_type):

# get logger instance.

self._logger = Util.get_logger('SPOT.INGEST.WRK.FLOW')

self._db_name = db_name

self._hdfs_app_path = hdfs_app_path

# read proxy configuration.

self._script_path = os.path.dirname(os.path.abspath(__file__))

conf_file = "{0}/ingest_conf.json".format(os.path.dirname(os.path.dirname(self._script_path)))

conf = json.loads(open(conf_file).read())

self._conf = conf["pipelines"][conf_type]

self._id = "spot-{0}-worker".format(conf_type)

self._process_opt = self._conf['process_opt']

self._local_staging = self._conf['local_staging']

self.kafka_consumer = kafka_consumer

# self._cursor = hive_engine.create_connection()

self._cursor = hive_engine

示例2: _initialize_members

# 需要导入模块: import json [as 别名]

# 或者: from json import loads [as 别名]

def _initialize_members(self,db_name,hdfs_app_path,kafka_consumer,conf_type,processes):

# get logger instance.

self._logger = Util.get_logger('SPOT.INGEST.WRK.PROXY')

self._db_name = db_name

self._hdfs_app_path = hdfs_app_path

self._kafka_consumer = kafka_consumer

# read proxy configuration.

self._script_path = os.path.dirname(os.path.abspath(__file__))

conf_file = "{0}/ingest_conf.json".format(os.path.dirname(os.path.dirname(self._script_path)))

conf = json.loads(open(conf_file).read())

self._spark_conf = conf["spark-streaming"]

self._conf = conf["pipelines"][conf_type]

self._processes = processes

示例3: _get_suspicious_details

# 需要导入模块: import json [as 别名]

# 或者: from json import loads [as 别名]

def _get_suspicious_details(self):

uri_list = []

iana_conf_file = "{0}/components/iana/iana_config.json".format(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

if os.path.isfile(iana_conf_file):

iana_config = json.loads(open(iana_conf_file).read())

proxy_iana = IanaTransform(iana_config["IANA"])

for conn in self._proxy_scores:

clientip = conn[self._conf["proxy_score_fields"]["clientip"]]

fulluri = conn[self._conf["proxy_score_fields"]["fulluri"]]

date=conn[self._conf["proxy_score_fields"]["p_date"]].split('-')

if len(date) == 3:

year=date[0]

month=date[1].zfill(2)

day=date[2].zfill(2)

hh=(conn[self._conf["proxy_score_fields"]["p_time"]].split(":"))[0]

self._get_proxy_details(fulluri,clientip,year,month,day,hh,proxy_iana)