一个dbapi 2.0接口和用于databricks交互集群的sqlalchemy方言。
databricks-dbapi的Python项目详细描述
一个围绕pyhive的薄包装器，用于创建到交互式数据块群集的DBAPI连接。
还为databricks交互集群提供sqlalchemy方言。
安装
使用pip安装：
pip install databricks-dbapi
对于sqlalchemy支持，请使用安装：
pip install databricks-dbapi[sqlalchemy]
用法
connect()函数返回一个pyhive配置单元连接对象，该对象在内部包装一个thrift连接。
使用databricks api令牌（推荐）：
importosfromdatabricks_dbapiimportdatabrickstoken=os.environ["DATABRICKS_TOKEN"]host=os.environ["DATABRICKS_HOST"]cluster=os.environ["DATABRICKS_CLUSTER"]connection=databricks.connect(host=host,cluster=cluster,token=token,)cursor=connection.cursor()cursor.execute("SELECT * FROM some_table LIMIT 100")print(cursor.fetchone())print(cursor.fetchall())
使用您的用户名和密码（不推荐）：
importosfromdatabricks_dbapiimportdatabricksuser=os.environ["DATABRICKS_USER"]password=os.environ["DATABRICKS_PASSWORD"]host=os.environ["DATABRICKS_HOST"]cluster=os.environ["DATABRICKS_CLUSTER"]connection=databricks.connect(host=host,cluster=cluster,user=user,password=password)cursor=connection.cursor()cursor.execute("SELECT * FROM some_table LIMIT 100")print(cursor.fetchone())print(cursor.fetchall())
在azure平台上连接，或使用http_path：
importosfromdatabricks_dbapiimportdatabrickstoken=os.environ["DATABRICKS_TOKEN"]host=os.environ["DATABRICKS_HOST"]http_path=os.environ["DATABRICKS_HTTP_PATH"]connection=databricks.connect(host=host,http_path=http_path,token=token,)cursor=connection.cursor()cursor.execute("SELECT * FROM some_table LIMIT 100")print(cursor.fetchone())print(cursor.fetchall())
pyhive连接还提供异步功能：
importosfromdatabricks_dbapiimportdatabricksfromTCLIService.ttypesimportTOperationStatetoken=os.environ["DATABRICKS_TOKEN"]host=os.environ["DATABRICKS_HOST"]cluster=os.environ["DATABRICKS_CLUSTER"]connection=databricks.connect(host=host,cluster=cluster,token=token,)cursor=connection.cursor()cursor.execute("SELECT * FROM some_table LIMIT 100",async_=True)status=cursor.poll().operationStatewhilestatusin(TOperationState.INITIALIZED_STATE,TOperationState.RUNNING_STATE):logs=cursor.fetch_logs()formessageinlogs:print(message)# If needed, an asynchronous query can be cancelled at any time with:# cursor.cancel()status=cursor.poll().operationStateprint(cursor.fetchall())
炼金术
安装databricks-dbapi包后，databricks+pyhive方言/驱动程序将注册到sqlalchemy。在传递引擎url时填写所需的信息。
fromsqlalchemyimport*fromsqlalchemy.engineimportcreate_enginefromsqlalchemy.schemaimport*# Standard Databricks with user + password# provide user, password, company name for url, database name, cluster nameengine=create_engine("databricks+pyhive://<user>:<password>@<companyname>.cloud.databricks.com:443/<database>",connect_args={"cluster":"<cluster>"})# Standard Databricks with token# provide token, company name for url, database name, cluster nameengine=create_engine("databricks+pyhive://token:<databricks_token>@<companyname>.cloud.databricks.com:443/<database>",connect_args={"cluster":"<cluster>"})# Azure Databricks with user + password# provide user, password, region for url, database name, http_path (with cluster name)engine=create_engine("databricks+pyhive://<user>:<password>@<region>.azuredatabricks.net:443/<database>",connect_args={"http_path":"<azure_databricks_http_path>"})# Azure Databricks with token# provide token, region for url, database name, http_path (with cluster name)engine=create_engine("databricks+pyhive://token:<databrickstoken>@<region>.azuredatabricks.net:443/<database>",connect_args={"http_path":"<azure_databricks_http_path>"})logs=Table("my_table",MetaData(bind=engine),autoload=True)print(select([func.count("*")],from_obj=logs).scalar())
有关主机名、群集名和http路径的详细信息，请参阅以下文档：
欢迎加入QQ群-->： 979659372
