Python 调用 ES、Solr、Phoenix的示例代码

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# *************************************
# @Time  : 2019/8/12
# @Author : Zhang Fan
# @Desc  : Library
# @File  : MyDatabases.py
# @Update : 2019/8/23
# *************************************
import elasticsearch
import phoenixdb
import pysolr
import pymysql

class MyELS(object):
  """
  ===================================================================
  =====================    MyELS    =========================
  ===================================================================
  """
  def __init__(self):
    self.els_conn = None

  def connect_to_els(self, host, port):
    """
    连接到ElasticSearch服务器.
    """
    self.els_conn = elasticsearch.Elasticsearch([{'host': host, 'port': port}])
    print('Executing : Connect To Elastic Search | %s' % self.els_conn)

  def get_els_data(self, query, index):
    """
    获取ElasticSearch数据
    """
    print('Executing : Search | %s' % query)
    try:
      rst = self.els_conn.search(index=index, q=query)
      return rst['hits']
    except Exception as e:
      print('Elastic Search Error | %s' % e)
      raise Exception(e)

class MyPhoenix(object):
  """
  ===================================================================
  =====================    MyPhoenix    ======================
  ===================================================================
  """
  def __init__(self):
    self.phoenix_conn = None
    self.phoenix_cursor = None

  def connect_to_phoenix(self, host, port=8765):
    """
    连接到phoenix服务器
    """
    address = 'http://{0}:{1}/'.format(host, port)
    print('Executing : Connect To Phoenix | %s' % address)
    self.phoenix_conn = phoenixdb.connect(address, autocommit=True)
    self.phoenix_cursor = self.phoenix_conn.cursor()

  def set_schema(self, sql, schema):
    """
    设置schema
    """
    pre_sub, sub, fol_sub = sql.upper().partition('FROM')
    fol_sub = ' ' + schema + '.' + fol_sub.strip()
    new_sql = ''.join([pre_sub, sub, fol_sub])
    return new_sql

  def execute_phoenix_sql(self, sql):
    """
    执行sql语句
    """
    # sql = self.set_schema(sql, schema)
    print('Executing : Execute | %s' % sql)
    self.phoenix_cursor.execute(sql)

  def get_from_phoenix(self, sql):
    """
    获取phoenix数据
    """
    # sql = self.set_schema(sql, schema)
    print('Executing : Query | %s' % sql)
    try:
      self.phoenix_cursor.execute(sql)
    except Exception as e:
      print('Phoenix Error | %s' % e)
      raise Exception(e)
    return self.phoenix_cursor.fetchall()

  def disconnect_from_phoenix(self):
    """
    断开phoenix连接
    """
    print('Executing : Disconnect From HBase')
    self.phoenix_cursor.close()
    self.phoenix_conn.close()

class MySolr(object):
  """
  ===================================================================
  =====================    MySolr    =========================
  ===================================================================
  """
  def __init__(self):
    self.solr_conn = None
    self.base_url = None

  def connect_to_solr(self, address, selector):
    """连接到solr服务器.
    """
    self.base_url = 'http://{0}/solr/{1}/'.format(address, selector)
    self.solr_conn = pysolr.Solr(self.base_url)
    print('Executing : Connect To Solr | %s' % self.base_url)

  def get_solr_data(self, query):
    """
    获取solr数据
    """
    results = list()
    print('Executing : Search | %s' % query)
    try:
      items = self.solr_conn.search(query)
      for item in items:
        results.append(item)
    except Exception as e:
      print('Solr Error | %s' % e)
      raise Exception(e)
    return results

  def add_solr_data(self, data):
    """
    添加solr数据
    """
    print('Executing : add | %s' % data)
    try:
      self.solr_conn.add([data])
      self.solr_conn.commit()
    except Exception as e:
      print('Solr Error | %s' % e)
      raise Exception(e)

  def del_solr_byId(self, data):
    """
    删除solr数据
    """
    print('Executing : del | %s' % data)
    try:
      self.solr_conn.delete(id=data)
      self.solr_conn.commit()
    except Exception as e:
      print('Solr Error | %s' % e)
      raise Exception(e)

if __name__ == '__main__':
  print('This is test.')
  ms = MySolr()
  me = MyELS()
  mp = MyPhoenix()

以上就是Python 调用 ES、Solr、Phoenix的示例代码的详细内容,更多关于Python 调用 ES、Solr、Phoenix的资料请关注我们其它相关文章!

(0)

相关推荐

  • python中使用ctypes调用so传参设置遇到的问题及解决方法

    问题 近日在做一组声纹聚类时,使用了另一团队同学开发的声纹距离算法.该算法对外提供的是一组so包,需要使用方自己去使用.在python中调用纯so包一般使用ctypes类库,用起来看起来简单但也有不少细节容易犯错.本次使用过程中,就遇到传参的问题. 目标so库中对外export的函数是大致如下的三个函数: void* create_handler(); int extract_feature(void* hander); bool destroy(void* handler); 这三个函数使用起

  • Python使用ctypes调用C/C++的方法

    python使用ctypes调用C/C++ 1. ctpes介绍 ctypes is a foreign function library for Python. It provides C compatible data types, and allows calling functions in DLLs or shared libraries. It can be used to wrap these libraries in pure Python. 官方文档地址: https://do

  • Python调用C语言的方法【基于ctypes模块】

    本文实例讲述了Python调用C语言的方法.分享给大家供大家参考,具体如下: Python中的ctypes模块可能是Python调用C方法中最简单的一种.ctypes模块提供了和C语言兼容的数据类型和函数来加载dll文件,因此在调用时不需对源文件做任何的修改.也正是如此奠定了这种方法的简单性. 示例如下 实现两数求和的C代码,保存为add.c //sample C file to add 2 numbers - int and floats #include <stdio.h> int add

  • 如何使用python的ctypes调用医保中心的dll动态库下载医保中心的账单

    需求:根据医保中心的文档和提供的dll动态库调用相关接口下载医保中心的账单. 文档:对调用dll动态库的描述,调用哪个dll文件,同时了解清楚调用这个dll文件中的哪个函数. 分析:结合文档及相关介绍弄清楚相关接口调用流程,从以上可以看出接口调用的是SiInterface.dll文件,然后先调用INIT函数进行初始化,然后再调用BUSINESS_HANDLE函数在医保局签到,然后在次调用BUSINESS_HANDLE函数下载账单,同时根据文档分析出每次调用函数的出入参.(具体的调用流程及每个函数

  • Python调用REST API接口的几种方式汇总

    相信做过自动化运维的同学都用过REST API接口来完成某些动作.API是一套成熟系统所必需的接口,可以被其他系统或脚本来调用,这也是自动化运维的必修课. 本文主要介绍python中调用REST API的几种方式,下面是python中会用到的库. - urllib2 - httplib2 - pycurl - requests urllib2 - Sample1 import urllib2, urllib github_url = 'https://api.github.com/user/re

  • python调用百度语音REST API

    本文实例为大家分享了python调用百度语音REST API的具体代码,供大家参考,具体内容如下 (百度的rest接口的部分网址发生了一定的变化,相关代码已更新) 百度通过 REST API 的方式给开发者提供一个通用的 HTTP 接口,基于该接口,开发者可以轻松的获得语音合成与语音识别能力.SDK中只提供了PHP.C和JAVA的相关样例,使用python也可以灵活的对端口进行调用,本文描述了简单使用Python调用百度语音识别服务 REST API 的简单样例. 1.语音识别与语音合成的调用

  • python使用ctypes库调用DLL动态链接库

    最近要使用python调用C++编译生成的DLL动态链接库,因此学习了一下ctypes库的基本使用. ctypes是一个用于Python的外部函数库,它提供C兼容的数据类型,并允许在DLL或共享库中调用函数. 一.Python调用DLL里面的导出函数 1.VS生成dll 1.1 新建动态链接库项目 1.2 在myTest.cpp中输入以下内容: // myTest.cpp : 定义 DLL 应用程序的导出函数. // #include "stdafx.h" #define DLLEXP

  • python使用ctypes调用扩展模块的实例方法

    楔子 我们知道python的执行效率不是很高,而且由于GIL的原因,导致python不能充分利用多核CPU.一般的解决方式是使用多进程,但是多进程开销比较大,而且进程之间的通信也会比较麻烦.因此在解决效率问题上,我们会把那些比较耗时的模块使用C或者C++编写,然后编译成动态链接库,Windows上面是dll,linux上面则是so,编译好之后,交给python去调用.而且通过扩展模块的方式还可以解决python的GIL的问题,因此如果想要利用多核,我们仍然可以通过扩展模块的方式. python如

  • Python如何通过subprocess调用adb命令详解

    前言 本文主要给大家介绍了关于使用Python通过subprocess调用adb命令,subprocess包主要功能是执行外部命令(相对Python而言).和shell类似. 换言之除了adb命令外,利用subprocess可以执行其他的命令,比如ls,cd等等. subprocess 可参考: https://docs.python.org/2/library/subprocess.html 在电脑上装好adb工具,配置好adb的环境变量,先确保shell中可以调用adb命令. 代码示例 Py

  • python调用百度REST API实现语音识别

    目前,语音识别,即将语音内容转换为文字的技术已经比较成熟,遥想当时锤子发布会上展示的讯飞输入法语音识别,着实让讯飞火了一把.由于此类语音识别需要采集大量的样本,才能达到一定的准确度,个人很难从零开始搭建.但是,许多拥有语音识别技术的公司,或多或少会提供一些API或者SDK供开发者使用,这样就把语音识别的门槛降到了一个很低的程度,只需几行代码即可实现.下面我介绍以下如何使用Python调用百度的REST API实现一个简单的语音识别. 注册账号,并成为开发者 打开 http://yuyin.bai

随机推荐