导入 cassandra 的行数不匹配 table（COPY 命令）

Question

我正在尝试使用 COPY 命令将 csv 文件转储到 cassandra table 中。但是，我的 csv 文件中的行数和 cassandra 中的行数不一致。

CSV 文件中的行数：49765（不包括 header）

cassandra 中的行数table:

cqlsh:test_df> select Count(*) from test_table;

 count
-------
 46982

(1 rows)

Warnings :
Aggregation query used without partition key

复制命令：

COPY test_table (column1,column2,column3) from 'temp.csv'  with delimiter = ',' and header = True;

错误：

Starting copy of test_df.test_bhavcopy with columns [symbol, instrument, expiry_dt, strike_pr, option_typ, open, high, low, close, settle_pr, contracts, val_inlakh, open_int, ch_in_oi, price_date, key].
Process ImportProcess-3:ate:    8387 rows/s; Avg. rate:    3937 rows/s
Traceback (most recent call last):
P rocess ImportProcess-2:
 File "X:\Anaconda\lib\multiprocessing\process.py", line 267, in _bootstrap
Traceback (most recent call last):
Process ImportProcess-1:
T raceback (most recent call last):
  File "X:\Anaconda\lib\multiprocessing\process.py", line 267, in _bootstrap
 File "X:\Anaconda\lib\multiprocessing\process.py", line 267, in _bootstrap
    self.run()
    File "X:\apache-cassandra-3.11.3\bin\..\pylib\cqlshlib\copyutil.py", line 2328, in run
   self.run()
   self.run()
   File "X:\apache-cassandra-3.11.3\bin\..\pylib\cqlshlib\copyutil.py", line 2328, in run
 File "X:\apache-cassandra-3.11.3\bin\..\pylib\cqlshlib\copyutil.py", line 2328, in run
    self.close()
  File "X:\apache-cassandra-3.11.3\bin\..\pylib\cqlshlib\copyutil.py", line 2332, in close
    self._session.cluster.shutdown()
      self.close()
 File "X:\apache-cassandra-3.11.3\bin\..\lib\cassandra-driver-internal-only-3.11.0-bb96859b.zip\cassandra-driver-3.11.0-bb96859b\cassandra\cluster.py", line 1259, in shutdown
   self.close()
   File "X:\apache-cassandra-3.11.3\bin\..\pylib\cqlshlib\copyutil.py", line 2332, in close
 File "X:\apache-cassandra-3.11.3\bin\..\pylib\cqlshlib\copyutil.py", line 2332, in close
     self._session.cluster.shutdown()
   self._session.cluster.shutdown()
   File "X:\apache-cassandra-3.11.3\bin\..\lib\cassandra-driver-internal-only-3.11.0-bb96859b.zip\cassandra-driver-3.11.0-bb96859b\cassandra\cluster.py", line 1259, in shutdown
 File "X:\apache-cassandra-3.11.3\bin\..\lib\cassandra-driver-internal-only-3.11.0-bb96859b.zip\cassandra-driver-3.11.0-bb96859b\cassandra\cluster.py", line 1259, in shutdown
    self.control_connection.shutdown()
  File "X:\apache-cassandra-3.11.3\bin\..\lib\cassandra-driver-internal-only-3.11.0-bb96859b.zip\cassandra-driver-3.11.0-bb96859b\cassandra\cluster.py", line 2850, in shutdown
    self._connection.close()
  File "X:\apache-cassandra-3.11.3\bin\..\lib\cassandra-driver-internal-only-3.11.0-bb96859b.zip\cassandra-driver-3.11.0-bb96859b\cassandra\io\asyncorereactor.py", line 373, in close
    AsyncoreConnection.create_timer(0, partial(asyncore.dispatcher.close, self))
  File "X:\apache-cassandra-3.11.3\bin\..\lib\cassandra-driver-internal-only-3.11.0-bb96859b.zip\cassandra-driver-3.11.0-bb96859b\cassandra\io\asyncorereactor.py", line 335, in create_timer
    cls._loop.add_timer(timer)
A ttributeError: 'NoneType' object has no attribute 'add_timer'
   self.control_connection.shutdown()
   File "X:\apache-cassandra-3.11.3\bin\..\lib\cassandra-driver-internal-only-3.11.0-bb96859b.zip\cassandra-driver-3.11.0-bb96859b\cassandra\cluster.py", line 2850, in shutdown
   self.control_connection.shutdown()
     self._connection.close()
 File "X:\apache-cassandra-3.11.3\bin\..\lib\cassandra-driver-internal-only-3.11.0-bb96859b.zip\cassandra-driver-3.11.0-bb96859b\cassandra\cluster.py", line 2850, in shutdown
   File "X:\apache-cassandra-3.11.3\bin\..\lib\cassandra-driver-internal-only-3.11.0-bb96859b.zip\cassandra-driver-3.11.0-bb96859b\cassandra\io\asyncorereactor.py", line 373, in close
   self._connection.close()
  File "X:\apache-cassandra-3.11.3\bin\..\lib\cassandra-driver-internal-only-3.11.0-bb96859b.zip\cassandra-driver-3.11.0-bb96859b\cassandra\io\asyncorereactor.py", line 373, in close
    AsyncoreConnection.create_timer(0, partial(asyncore.dispatcher.close, self))
     AsyncoreConnection.create_timer(0, partial(asyncore.dispatcher.close, self))
 File "X:\apache-cassandra-3.11.3\bin\..\lib\cassandra-driver-internal-only-3.11.0-bb96859b.zip\cassandra-driver-3.11.0-bb96859b\cassandra\io\asyncorereactor.py", line 335, in create_timer
   File "X:\apache-cassandra-3.11.3\bin\..\lib\cassandra-driver-internal-only-3.11.0-bb96859b.zip\cassandra-driver-3.11.0-bb96859b\cassandra\io\asyncorereactor.py", line 335, in create_timer
   cls._loop.add_timer(timer)
 A   cls._loop.add_timer(timer)
ttributeError: 'NoneType' object has no attribute 'add_timer'
AttributeError: 'NoneType' object has no attribute 'add_timer'
Processed: 49765 rows; Rate:    4193 rows/s; Avg. rate:    3906 rows/s
49765 rows imported from 1 files in 12.742 seconds (0 skipped).

可能是因为这个错误。

Answer 1

找到一个修复：我在

中编辑了我的 asyncorereactor.py

cassandra-driver-internal-only-3.11.0-bb96859b.zip/cassandra-driver-3.11.0-bb96859b/cassandra/io/asyncorereactor.py

从 AsyncoreConnection.create_timer() 到 self.create_timer()，如本 post

中所建议

https://datastax-oss.atlassian.net/browse/PYTHON-862?page=com.atlassian.jira.plugin.system.issuetabpanels%3Aall-tabpanel

导入 cassandra 的行数不匹配 table（COPY 命令）

Mismatch in number of rows imported into cassandra table (COPY command)

cql

cassandra