我希望由postgresql作为生产者，kafka监听数据库变动，当数据库有修改时生产者会生产数据变动的json数据，我希望这个数据只包含有效负载传给消费者，怎么做？

要实现PostgreSQL作为生产者，Kafka监听数据库变动，并且只传递有效负载数据给消费者，您可以按照以下步骤进行设置：

1. 首先，您需要在PostgreSQL数据库中启用逻辑复制。在 `postgresql.conf` 配置文件中设置 `wal_level` 参数为 `logical`。

2. 创建逻辑复制槽，用于从数据库中捕获变更数据。您可以使用以下命令创建槽：

   SELECT * FROM pg_create_logical_replication_slot('my_slot', 'wal2json');

3. 在Kafka上配置并启动Kafka Connect的Debezium插件。Debezium插件可以监听PostgreSQL数据库的变动并将其转发到Kafka主题。
请参考Debezium文档进行详细配置和启动。

4. 在Kafka Connect的配置文件中，配置Debezium连接器以连接到PostgreSQL数据库和Kafka集群。设置以下参数：
- `plugin.name=io.debezium.connector.postgresql.PostgresConnector`
- `database.hostname=your_postgresql_host`
- `database.port=your_postgresql_port`
- `database.user=your_postgresql_username`
- `database.password=your_postgresql_password`
- `database.dbname=your_postgresql_database_name`
- `database.server.name=your_kafka_server_name`
- `database.slot.name=my_slot`
- `table.include.list=your_table_name`

请根据您的实际情况修改上述参数。

5. 根据您的需求，配置Debezium连接器的转换器。设置以下参数：
- `key.converter=org.apache.kafka.connect.json.JsonConverter`
- `value.converter=org.apache.kafka.connect.json.JsonConverter`
- `key.converter.schemas.enable=false`
- `value.converter.schemas.enable=false`

这将确保只传递有效负载数据，而不包含模式信息。

6. 启动Kafka Connect并验证是否成功监听并将变动数据传递到Kafka主题。消费者可以从该主题订阅数据。

通过这些步骤，您可以实现PostgreSQL作为生产者，Kafka监听数据库变动，并且只传递有效负载数据给消费者。

请注意，具体的设置和配置可能因您使用的工具和版本而有所不同。您需要根据您的实际情况进行适当的调整和配置。

希望这些信息对您有所帮助！如果您有任何其他问题，请随时提问。