为什么插入查询会导致我的数据流管道停止处理数据库插入？

Question

我有一个 Google Cloud Dataflow 管道（流），它全天处理大量插入并将它们插入 Cloud SQL。我正在尝试用现有数据填充另一个表，但是当我尝试插入这些表时，它会导致管道停止插入。第二次我停止执行查询，插入继续插入没有问题。

我在我的数据库上尝试了其他查询，但只有插入会导致它停止处理插入。

我的插入查询是：

INSERT INTO t2 (c1, c2, firstUsed, lastUsed)
SELECT DISTINCT c3, c4, MIN(created), MAX(created) FROM t1 WHERE created 
<= '2018-12-01 23:59:59' GROUP BY c3, c4
ON DUPLICATE KEY UPDATE lastUsed = VALUES(lastUsed)

我在 Google Cloud 上收到的异常是：

Processing stuck in step ParDo(ProcessDatabaseEvent) for at least 05m00s without outputting or completing in state process
  at java.net.SocketInputStream.socketRead0(Native Method)
  at java.net.SocketInputStream.socketRead(SocketInputStream.java:116)
  at java.net.SocketInputStream.read(SocketInputStream.java:170)
  at java.net.SocketInputStream.read(SocketInputStream.java:141)
  at com.mysql.cj.protocol.ReadAheadInputStream.fill(ReadAheadInputStream.java:107)
  at com.mysql.cj.protocol.ReadAheadInputStream.readFromUnderlyingStreamIfNecessary(ReadAheadInputStream.java:150)
  at com.mysql.cj.protocol.ReadAheadInputStream.read(ReadAheadInputStream.java:180)
  at java.io.FilterInputStream.read(FilterInputStream.java:133)
  at com.mysql.cj.protocol.FullReadInputStream.readFully(FullReadInputStream.java:64)
  at com.mysql.cj.protocol.a.SimplePacketReader.readHeader(SimplePacketReader.java:63)
  at com.mysql.cj.protocol.a.SimplePacketReader.readHeader(SimplePacketReader.java:45)
  at com.mysql.cj.protocol.a.TimeTrackingPacketReader.readHeader(TimeTrackingPacketReader.java:52)
  at com.mysql.cj.protocol.a.TimeTrackingPacketReader.readHeader(TimeTrackingPacketReader.java:41)
  at com.mysql.cj.protocol.a.MultiPacketReader.readHeader(MultiPacketReader.java:54)
  at com.mysql.cj.protocol.a.MultiPacketReader.readHeader(MultiPacketReader.java:44)
  at com.mysql.cj.protocol.a.NativeProtocol.readMessage(NativeProtocol.java:557)
  at com.mysql.cj.protocol.a.NativeProtocol.checkErrorMessage(NativeProtocol.java:735)
  at com.mysql.cj.protocol.a.NativeProtocol.sendCommand(NativeProtocol.java:674)
  at com.mysql.cj.protocol.a.NativeProtocol.sendQueryPacket(NativeProtocol.java:966)
  at com.mysql.cj.NativeSession.execSQL(NativeSession.java:1165)
  at com.mysql.cj.jdbc.ClientPreparedStatement.executeInternal(ClientPreparedStatement.java:937)
  at com.mysql.cj.jdbc.ClientPreparedStatement.executeUpdateInternal(ClientPreparedStatement.java:1116)
  at com.mysql.cj.jdbc.ClientPreparedStatement.executeUpdateInternal(ClientPreparedStatement.java:1066)
  at com.mysql.cj.jdbc.ClientPreparedStatement.executeLargeUpdate(ClientPreparedStatement.java:1396)
  at com.mysql.cj.jdbc.ClientPreparedStatement.executeUpdate(ClientPreparedStatement.java:1051)
  at mypardo.processElement(StatsCollector.java:173)
  at mypardo.invokeProcessElement(Unknown Source)
timestamp 2019-04-04T17:40:01.724Z
logger com.google.cloud.dataflow.worker.DataflowOperationContext
severity WARNING
worker myworker
step ParDo(ProcessDatabaseEvent)
thread 34

任何帮助将不胜感激！

编辑：

添加数据库架构

t1：（大约 1000 万条记录）

c1          varchar(255)  NO   PRI NULL    
c2          varchar(255)  NO   PRI NULL     
c3          varchar(255)  YES  NULL    
c4          varchar(255)  YES  NULL    
c5          varchar(255)  YES  NULL    
c6          varchar(255)  YES  NULL    
created     datetime      YES  NULL

t2:

c1         varchar(255) NO   PRI NULL    
c2         varchar(255) NO   PRI NULL    
firstUsed  datetime     YES      NULL    
lastUsed   datetime     YES      NULL    

Answer 1

如果你使用 group by 你不需要 distinct 并且你不应该在 ON DUPLICATE 类中使用 values(lastUsed) 而是 t1.lastUsed

  INSERT INTO t2 (c1, c2, firstUsed, lastUsed)
  SELECT c3, c4, MIN(created), MAX(created) 
  FROM t1 
  WHERE created <= '2018-12-01 23:59:59' 
  GROUP BY c3, c4
  ON DUPLICATE KEY UPDATE lastUsed = t1.lastUsed