FreeTDSとPyODBCランダム接続がサーバーから切断される

Question

何らかの理由で、次のスタックトレースは、プロセスの開始から約16時間後に最初に表示され、その後1時間または2時間ごとに表示されます。毎日、毎時、そして数分おきに他のサーバー上に他のジョブがあり、このスタックトレースを取得しません。私が見ることができる唯一の違いは、このコードはスケジューラごとに複数のジョブを持っているのに対し、他のスケジ

スタックトレース：

Traceback (most recent call last): 
    File "app.py", line 24, in _run_client 
    h.run_hourly(start, end) 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/foo/client/bar/helper.py", line 153, in run_hourly 
    _run(_HOURLY_REQUEST, frequency, start, end, duration) 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/foo/client/bar/helper.py", line 117, in _run 
    bars = dr.get_bars(frequency) 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/foo/client/bar/data_requests.py", line 35, in get_bars 
    df = pd.read_sql(query, _ENGINE, params=params) 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/pandas/io/sql.py", line 415, in read_sql 
chunksize=chunksize) 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/pandas/io/sql.py", line 1084, in read_query 
    result = self.execute(*args) 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/pandas/io/sql.py", line 975, in execute 
    return self.connectable.execute(*args, **kwargs) 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 2055, in execute 
    return connection.execute(statement, *multiparams, **params) 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 945, in execute 
    return meth(self, multiparams, params) 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/sqlalchemy/sql/elements.py", line 263, in _execute_on_connection 
    return connection._execute_clauseelement(self, multiparams, params) 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 1053, in _execute_clauseelement 
compiled_sql, distilled_params 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 1189, in _execute_context 
context) 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 1386, in _handle_dbapi_exception 
    self._autorollback() 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 824, in _autorollback 
    self._root._rollback_impl() 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 703, in _rollback_impl 
    self._handle_dbapi_exception(e, None, None, None, None) 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 1315, in _handle_dbapi_exception 
    exc_info 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/sqlalchemy/util/compat.py", line 202, in raise_from_cause 
    reraise(type(exception), exception, tb=exc_tb, cause=cause) 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/sqlalchemy/engine/base.py", line 701, in _rollback_impl 
    self.engine.dialect.do_rollback(self.connection) 
    File "/home/foo/anaconda2/lib/python2.7/site-packages/sqlalchemy/engine/default.py", line 439, in do_rollback 
    dbapi_connection.rollback() 
DBAPIError: (pyodbc.Error) ('08S01', '[08S01] [FreeTDS][SQL Server]Write to the server failed (20006) (SQLEndTran)') 

下記のバージョンに関して内のサーバーの試合。

バージョン：
OS：Ubuntuの16.04.1
FreeTDSのパッケージバージョン：FreeTDSの共通：0.91-6.1build1、tdsodbc：AMD64：0.91-6.1build1 PyODBC：3.0.10
SQL錬金術：1.1。 4

TSQL出力：

Compile-time settings (established with the "configure" script) 
         Version: freetds v0.91 
     freetds.conf directory: /etc/freetds 
MS db-lib source compatibility: no 
    Sybase binary compatibility: yes 
        Thread safety: yes 
        iconv library: yes 
        TDS version: 4.2 
          iODBC: no 
         unixodbc: yes 
      SSPI "trusted" logins: no 
         Kerberos: yes 

テストコード：

app.py

from datetime import datetime, timedelta 
from dateutil import relativedelta 
import traceback 

from apscheduler.schedulers.blocking import BlockingScheduler 
from bar import helper as h 


def _run_client(resolution): 
    try: 
     if resolution == "hourly": 
      # Had to create a temporary variable to make native datetimes 
      t = datetime.utcnow() - timedelta(hours=1) 
      end = datetime(t.year, t.month, t.day, t.hour) 
      start = end - timedelta(hours=1) 
      h.run_hourly(start, end) 
     elif resolution == "daily": 
      # Had to create a temporary variable to make native datetimes 
      t = datetime.utcnow().date() - timedelta(days=1) 
      end = datetime(t.year, t.month, t.day) 
      start = end - timedelta(days=1) 
      h.run_daily(start, end) 
     else: 
      # Had to create a temporary variable to make native datetimes 
      t = datetime.utcnow().date().replace(
       day=1) - relativedelta.relativedelta(months=1) 
      end = datetime(t.year, t.month, t.day) 
      start = end - relativedelta.relativedelta(months=1) 
      h.run_monthly(start, end) 
    except: 
     print "Current run failed:\n%s" % traceback.format_exc() 


def _get_hourly_job(sched): 
    args = ["hourly"] 
    job = sched.add_job(_run_client, args=args, trigger="cron", hour="*", minute="0") 
    return job 


def _get_daily_job(sched): 
    args = ["daily"] 
    job = sched.add_job(_run_client, args=args, trigger="cron", hour="4", minute="0") 
    return job 


def _get_monthly_job(sched): 
    args = ["monthly"] 
    job = sched.add_job(_run_client, args=args, trigger="cron", day="1", hour="0", minute="0") 
    return job 

if __name__ == "__main__": 
    sched = BlockingScheduler() 
    hourly_job = _get_hourly_job(sched) 
    daily_job = _get_daily_job(sched) 
    monthly_job = _get_monthly_job(sched) 

    try: 
     sched.start() 
    except: 
     # Remove the jobs from memory since they finished 
     hourly_job.remove() 
     daily_job.remove() 
     monthly_job.remove() 

     sched.shutdown() 

helper.py

from datetime import timedelta 

import data_requests as dr 

_HOURLY_REQUEST = Foo() 
_HOURLY_REQUEST.resolution = "hourly" 

_DAILY_REQUEST = Foo() 
_DAILY_REQUEST.resolution = "daily" 

_MONTHLY_REQUEST = Foo() 
_MONTHLY_REQUEST.resolution = "monthly" 


def _run(request, frequency, start, end, duration): 
    bars = dr.get_bars(frequency) 

    if bars.empty: 
     return None 
    print "Bars = %i" % len(bars) 


def run_daily(start, end): 
    frequency = 86400 
    duration = timedelta(hours=4) 
    _run(_DAILY_REQUEST, frequency, start, end, duration) 


def run_hourly(start, end): 
    frequency = 3600 
    duration = timedelta(minutes=30) 
    _run(_HOURLY_REQUEST, frequency, start, end, duration) 


def run_monthly(start, end): 
    frequency = 1209600 
    duration = timedelta(days=1) 
    _run(_MONTHLY_REQUEST, frequency, start, end, duration) 

data_requests.py

import pandas as pd 
from sqlalchemy import create_engine, exc 
from sqlalchemy.sql import text 

_DB = "mssql+pyodbc://foo@stg-foo:FooBar@stg-foo.database.secure.windows.net:1433/foo?driver=/usr/lib/x86_64-linux-gnu/odbc/libtdsodbc.so&tds_version=7.2" 
_ENGINE = create_engine(_DB) 


def get_bars(frequency): 
    query = text("""SELECT h.foo_id, s.id AS bar_id, u.timezone 
       FROM foo h 
       INNER JOIN bar s ON h.foo_id = s.foo_id 
       AND s.frequency <= :frequency 
       INNER JOIN test u ON u.id = h.test_id""") 
    params = { 
     "frequency": int(frequency) 
    } 

    try: 
     df = pd.read_sql(query, _ENGINE, params=params) 
    except exc.DBAPIError, e: 
     # If connection is invalid (e.g. database restarted) execute the query 
     # again 
     if e.connection_invalidated: 
      df = pd.read_sql(query, _ENGINE, params=params) 
     else: 
      raise e 

    return df 

は私がpymssqlを試してみましたが、それは、最新バージョン（2.1.3、バージョン1.00.9をFreeTDSの）でセグメンテーションフォールト、 pyodbcはそうではありません。私はこれがドライバーの問題だと思うが、問題を回避しようとする問題がある。

さらに、私はSQL Alchemyのドキュメンテーションに従って接続切断を処理しましたが、スタックトレースを見れば、ロジックには当てられていないようです。代わりに、elseブロックにヒットし、エラーを発生させます。

Answer 1

エラーの原因が見つかりました。結局のところコーディング上の問題だったようです。わかるには長い時間がかかりましたが、パンダで電話をかける方法と関係していました。どうやら、エンジンインスタンスを渡すときにpandasは接続を閉じるように見えません。問題を解決するために使用したコードは次のとおりです。

def _get_df(query, params=None): 
    try: 
     with _ENGINE.begin() as conn: 
      df = pd.read_sql(query, conn, params=params) 
    except exc.DBAPIError, e: 
     # If connection is invalid (e.g. database restarted) execute the query 
     # again 
     if e.connection_invalidated: 
      with _ENGINE.begin() as conn: 
       df = pd.read_sql(query, conn, params=params) 
     else: 
      raise e 
    return df