Apache Beam: java.lang.illegalstateException при чтении из таблицы MSSQL ⇐ JAVA

[Anonymous]

У меня есть конвейер пучка, который считывает из таблицы MSSQL, используя простой запрос: < /p>
return "SELECT "
+ "U.ID as userid, "
+ "U.firstname as firstname, "
+ "U.lastname as lastname, "
+ "email as email, "
+ "U.IP as ip "
+ "FROM users U WITH (NOLOCK) "
+ "OPTION (MAXDOP 4)";
< /code>
Моя (упрощенная) таблица пользователей выглядит так: < /p>
CREATE TABLE users (
[ID] INT IDENTITY (1000000, 1) NOT NULL,
[Firstname] NVARCHAR (30) CONSTRAINT [DF_iddb_Nom] DEFAULT ('') NOT NULL,
[Lastname] NVARCHAR (30) CONSTRAINT [DF_iddb_Prenom] DEFAULT ('') NOT NULL,
[Email] NVARCHAR (40) NOT NULL,
[IP] VARCHAR (15) NULL
)
< /code>
Мой процессор: < /p>
public static class DataReadFn extends DoFn {

@ProcessElement
public void processElement(ProcessContext c) {
Row row = c.element();

User user = new User()
.setUserId(format(String.valueOf(row.getInt32("userid"))))
.setFirstName(format(row.getString("firstname")))
.setLastName(format(row.getString("lastname")))
.setEmail(format(row.getString(“email”))
.setIp(format(row.getString("ip"))); // last

c.output(record);
}

private String format(String str) {
return str == null ? "" : str.trim();
}

}
< /code>
Когда я выполняю этот конвейер на Dataflow, я получаю следующее исключение только тогда, когда я включаю Lastloginip в запрос. Без LASTLOGINIP он работает нормально. < /P>
java.lang.IllegalStateException
at org.apache.beam.sdk.util.Preconditions.checkStateNotNull(Preconditions.java:452)
at org.apache.beam.sdk.io.jdbc.SchemaUtil.lambda$createLogicalTypeExtractor$83184fac$1(SchemaUtil.java:307)
at org.apache.beam.sdk.io.jdbc.SchemaUtil$BeamRowMapper.mapRow(SchemaUtil.java:380)
at org.apache.beam.sdk.io.jdbc.SchemaUtil$BeamRowMapper.mapRow(SchemaUtil.java:358)
at org.apache.beam.sdk.io.jdbc.JdbcIO$ReadFn.processElement(JdbcIO.java:1498)
< /code>
Это, вероятно, как -то связано с фактом, что Lastloginip является нулевым, но не может точно сказать. Есть идеи? Спасибо.

Подробнее здесь: https://stackoverflow.com/questions/763 ... ssql-table