# CSV field to TimeStamp

**URL:** <https://forum.confluent.io/t/csv-field-to-timestamp/1315>\
**Category:** Kafka Connect\
**Created:** [26 March 2021 10:32 UTC](https://forum.confluent.io/t/csv-field-to-timestamp/1315 "2021-03-26T10:32:27Z")\
**Posts on this page:** 8\
**Page:** 1

<div class="post-metadata">

**Author:** ![rvaneynd](https://avatars.discourse-cdn.com/v4/letter/r/7ea924/32.png) [@rvaneynd](https://forum.confluent.io/u/rvaneynd)\
**Post date:** [26 March 2021 10:32 UTC](https://forum.confluent.io/t/csv-field-to-timestamp/1315/1 "2021-03-26T10:32:27Z")

</div>

I am trying to load a CSV datetime field in to Kafka as a timestamp, using transforms and org.apache.kafka.connect.transforms.TimestampConverter$Value, but no luck.

The CSV field looks like this:

```auto
Fieldname: EventTime
Value: 08/02/2021 10:00

```

I tried this, but without success:

```auto
transforms = eventTimeValue
transforms.eventTimeValue.type=org.apache.kafka.connect.transforms.TimestampConverter$Value
transforms.eventTimeValue.target.type=Timestamp
transforms.eventTimeValue.format="dd/MM/yyyy' 'HH:mm"
transforms.eventTimeValue.field=EventTime

```

Any ideas on why this is not working?

---

<div class="post-metadata">

**Author:** ![rmoff](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/rmoff/32/3_2.png) [@rmoff](https://forum.confluent.io/u/rmoff)\
**Post date:** [26 March 2021 11:38 UTC](https://forum.confluent.io/t/csv-field-to-timestamp/1315/2 "2021-03-26T11:38:03Z")

</div>

What’s your full connector config? When you say “no luck” do you mean you get an error?

---

<div class="post-metadata">

**Author:** ![rvaneynd](https://avatars.discourse-cdn.com/v4/letter/r/7ea924/32.png) [@rvaneynd](https://forum.confluent.io/u/rvaneynd)\
**Post date:** [26 March 2021 12:00 UTC](https://forum.confluent.io/t/csv-field-to-timestamp/1315/3 "2021-03-26T12:00:12Z")

</div>

Robin, thank you for the quick response.

The full config is this.

```auto
name=Occupancy-Sftp
kafka.topic=Occupancy3
tasks.max=1
connector.class=io.confluent.connect.sftp.SftpCsvSourceConnector
cleanup.policy=MOVE
behavior.on.error=LOG
input.path=/Occupancy
error.path=/Occupancy/Error
finished.path=/Occupancy/Finished
input.file.pattern=.*\\.csv
sftp.username=<Username>
sftp.password=<Password>
sftp.host=<IP>
sftp.port=<Port>
csv.first.row.as.header=true
schema.generation.enabled=true
schema.generation.key.fields=EventTime
parser.timestamp.date.formats=dd/MM/yyyy' 'HH:mm 

transforms = eventTimeValue
transforms.eventTimeValue.type=org.apache.kafka.connect.transforms.TimestampConverter$Value
transforms.eventTimeValue.target.type=Timestamp
transforms.eventTimeValue.format=dd/MM/yyyy' 'HH:mm
transforms.eventTimeValue.field=EventTime

```

This is the data as put in the Topic:

```json
{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "string",
        "optional": true,
        "field": "EventTime"
      },
      {
        "type": "string",
        "optional": true,
        "field": "SourceEntityDescription"
      },
      {
        "type": "string",
        "optional": true,
        "field": "EventType"
      },
      {
        "type": "string",
        "optional": true,
        "field": "TokenNumber"
      }
    ],
    "optional": false,
    "name": "defaultvalueschemaname"
  },
  "payload": {
    "EventTime": "08/02/2021 11:57",
    "SourceEntityDescription": "Bla Bla Bla",
    "EventType": "Valid Access",
    "TokenNumber": "1234567890"
  }
}

```

---

<div class="post-metadata">

**Author:** ![rmoff](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/rmoff/32/3_2.png) [@rmoff](https://forum.confluent.io/u/rmoff)\
**Post date:** [26 March 2021 12:56 UTC](https://forum.confluent.io/t/csv-field-to-timestamp/1315/4 "2021-03-26T12:56:48Z")

</div>

I’m not familiar with this connector, but looking at [the docs](https://docs.confluent.io/kafka-connect-sftp/current/source-connector/csv_source_connector.html#timestamps) I think you need to specify

```auto
timestamp.mode=FIELD
timestamp.field=EventTime

```

This should then pick up the field as the timestamp to use as the Kafka message’s timestamp.

This isn’t the same as setting the datatype of the field itself. For that you need to work with the [schema generation](https://docs.confluent.io/kafka-connect-sftp/current/source-connector/csv_source_connector.html#schema-generation). You’ve set `schema.generation.key.fields` but I’m not sure if you want the timestamp as the message key?

If you’re trying to get the timestamp as a timestamp in the value of the message when it’s written to the topic then I would use an [explicit schema that you declare yourself](https://docs.confluent.io/kafka-connect-sftp/current/source-connector/csv_source_connector.html#schema) in `value.schema`

---

<div class="post-metadata">

**Author:** ![rvaneynd](https://avatars.discourse-cdn.com/v4/letter/r/7ea924/32.png) [@rvaneynd](https://forum.confluent.io/u/rvaneynd)\
**Post date:** [26 March 2021 15:55 UTC](https://forum.confluent.io/t/csv-field-to-timestamp/1315/5 "2021-03-26T15:55:29Z")

</div>

Robin, thank you.

I tried that using this config:

```auto
name=Havas-Occupancy-Sftp
kafka.topic=Havas-Occupancy3
tasks.max=1
connector.class=io.confluent.connect.sftp.SftpCsvSourceConnector
cleanup.policy=MOVE
behavior.on.error=LOG
input.path=/Occupancy/TestData
error.path=/Occupancy/TestData/Error
finished.path=/Occupancy/TestData/Finished
input.file.pattern=.*\\.csv
sftp.username=<Username>
sftp.password=<Password>
sftp.host=<IP>
sftp.port=<Port>
csv.first.row.as.header=true
schema.generation.enabled=false
key.schema={\"name\" :\"OccupancyNameSchema2\",\"type\" : \"STRUCT\",\"isOptional\" : true, \"fieldSchemas\" : {\"EventTime\" : {\"type\" : \"STRING\", \"isOptional\" : true}, \"SourceEntityDescription\" : {\"type\" : \"STRING\",\"isOptional\" : true}, \"EventType\" : {\"type\" : \"STRING\",\"isOptional\" : true}, \"TokenNumber\" : {\"type\" : \"STRING\",\"isOptional\" : true}}}
value.schema={\"name\" : \"OccupancyValueSchema2\",\"type\" : \"STRUCT\",\"isOptional\" : true, \"fieldSchemas\" : {\"EventTime\" : {\"name\":"\org.apache.kafka.connect.data.Timestamp\",\"type\":\"INT64\",\"version\":1,\"isOptional\": true}, \"SourceEntityDescription\" : {\"type\" : \"STRING\",\"isOptional\" : true}, \"EventType\" : {\"type\" : \"STRING\",\"isOptional\" : true}, \"TokenNumber\" : {\"type\" : \"STRING\",\"isOptional\" : true}}}

parser.timestamp.date.formats=[dd/MM/yyyy' 'HH:mm]

transforms = eventTimeValue
transforms.eventTimeValue.type=org.apache.kafka.connect.transforms.TimestampConverter$Value
transforms.eventTimeValue.target.type=Timestamp
transforms.eventTimeValue.format=dd/MM/yyyy' 'HH:mm
transforms.eventTimeValue.field=EventTime

```

Schema is:

```auto
key.schema=
{
	\"name\" :\"OccupancyNameSchema2\",\"type\" : \"STRUCT\",\"isOptional\" : true, 
	\"fieldSchemas\" : 
	{
		\"EventTime\" : {\"type\" : \"STRING\", \"isOptional\" : true},
		\"SourceEntityDescription\" : {\"type\" : \"STRING\",\"isOptional\" : true}, 
		\"EventType\" : {\"type\" : \"STRING\",\"isOptional\" : true}, 
		\"TokenNumber\" : {\"type\" : \"STRING\",\"isOptional\" : true}
	}
}

```

```auto
value.schema=
{
	\"name\" : \"OccupancyValueSchema2\",\"type\" : \"STRUCT\",\"isOptional\" : true, 
	\"fieldSchemas\" : 
	{
		\"EventTime\" : {\"name\":"\org.apache.kafka.connect.data.Timestamp\",\"type\":\"INT64\",\"version\":1,\"isOptional\": true},
		\"SourceEntityDescription\" : {\"type\" : \"STRING\",\"isOptional\" : true},
		\"EventType\" : {\"type\" : \"STRING\",\"isOptional\" : true},
		\"TokenNumber\" : {\"type\" : \"STRING\",\"isOptional\" : true}
	}
}

```

The response is now:

```auto
{
	"schema":
	{
		"type":"struct","fields":
		[
		{"type":"int64","optional":true,"name":"org.apache.kafka.connect.data.Timestamp","version":1,"field":"EventTime"},
		{"type":"string","optional":true,"field":"SourceEntityDescription"},
		{"type":"string","optional":true,"field":"EventType"},
		{"type":"string","optional":true,"field":"TokenNumber"}
		],
		"optional":true,"name":"HavasOccupancyValueSchema2"
	},
	"payload":
	{
		"EventTime":null,
		"SourceEntityDescription":"Bla Bla Bla",
		"EventType":"Valid Access",
		"TokenNumber":"1234567890"
	}
}

```

Made progress as the field is now defined as Timestamp but for some reason the transformation does not do what I expect it should do.  
I had a look at the definitions for [SimpleDateFormat](https://docs.confluent.io/platform/current/connect/transforms/timestampconverter.html), and I believe it is correct.

I tested the date time format using this website: [http://www.sdfonlinetester.info](http://www.sdfonlinetester.info) and looks correct.

---

<div class="post-metadata">

**Author:** ![rvaneynd](https://avatars.discourse-cdn.com/v4/letter/r/7ea924/32.png) [@rvaneynd](https://forum.confluent.io/u/rvaneynd)\
**Post date:** [27 March 2021 10:28 UTC](https://forum.confluent.io/t/csv-field-to-timestamp/1315/6 "2021-03-27T10:28:48Z")

</div>

Update…

Made progress. Discovered there is most likely a bug in the SFTP connector somewhere.  
If I let it generate a schema itself the data pull returns the EventTime field, no problem. But whatever I try with a custom schema the EventTime is always null.

Then I received updated CSV files with a new date time format for the EventTime field  
yyyy-MM-dd HH:mm:ss, and no problem pulling in the EventTime field and performing the transformation on it.

The previous datetime format was dd/MM/yyyy HH:mm, and I think the code has a problem with the / when a manual schema is defined.

As this source connector “io.confluent.connect.sftp.SftpCsvSourceConnector” is being developed by Confluent, how do I go about reporting this possible bug?

Robin, thank you for your help.

---

<div class="post-metadata">

**Author:** ![rmoff](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/rmoff/32/3_2.png) [@rmoff](https://forum.confluent.io/u/rmoff)\
**Post date:** [28 March 2021 11:42 UTC](https://forum.confluent.io/t/csv-field-to-timestamp/1315/7 "2021-03-28T11:42:04Z")

</div>

@nathan.nam any suggestions?☝

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/1X/c49438c90c9df282e9996fdf6971be890c71b65a.svg) [@system](https://forum.confluent.io/u/system)\
**Post date:** [12 April 2021 15:27 UTC](https://forum.confluent.io/t/csv-field-to-timestamp/1315/8 "2021-04-12T15:27:25Z")

</div>

This topic was automatically closed 7 days after the last reply. New replies are no longer allowed.