Произвольное соединение FreeTDS и PyODBC отключается от сервера

Question

Произвольное соединение FreeTDS и PyODBC отключается от сервера

По какой-то причине следующая трассировка стека сначала появляется примерно через 16 часов или дольше после запуска процесса, а затем появляется каждый час или два после. У меня есть другие задания на других серверах, которые выполняются ежедневно, ежечасно и каждые несколько минут, и они не получают эту трассировку стека. Единственное различие, которое я вижу, состоит в том, что у этого кода есть несколько заданий на планировщик, тогда как у других нет.

Трассировки стека:

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)')

Серверы совпадают в отношении версий, упомянутых ниже.

Версии:
ОС: Ubuntu 16.04.1
Версии пакета FreeTDS: freetds-common: 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, но это segfaults с последней версией (2.1.3, freetds версия 1.00.9), тогда как pyodbc нет. Я думаю, что это проблема с драйверами, но у меня проблемы, пытаясь обойти это.

Кроме того, я следовал документации SQL Alchemy о том, как обрабатывать разъединения соединений, но, глядя на трассировку стека, похоже, что это не соответствует логике. Вместо этого он попадает в блок else и вызывает ошибку.

1

pandas python-2.7 sqlalchemy pyodbc freetds

Источник

user6551407 14 дек '16 в 13:16

1 ответ

Решение

Другие вопросы по тегам pandas python-2.7 sqlalchemy pyodbc freetds

user6551407 14 дек '16 в 17:49 2016-12-14 17:49 · Accepted Answer · 2016-12-14 17:49

Я нашел причину ошибки. Казалось, это проблема кодирования в конце концов. Потребовалось много времени, чтобы понять, но это было связано с тем, как я звонил с пандами. По-видимому, панды не закрывают соединение при прохождении экземпляра движка. Вот код, который я использовал для решения проблемы.

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