您可以使用javaapi和maven依赖项（kafka和zookeeper）来创建kafka主题，如下所示。您可以从提交spark应用程序的代码中调用代码。

<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.3</version>
</dependency>

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.10</artifactId>
    <version>0.8.2.1</version>
</dependency>

import java.util.Properties;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkInterruptedException;
import kafka.utils.ZKStringSerializer$;
import kafka.admin.AdminUtils;

public final class KafkaUtils {
    public static void main(String[] args) throws Exception {       
        KafkaUtils.createTopic("x.x.x.x:2181,y.y.y.y:2181", "topicName", 1, 0, new Properties());       
    }

    public static void createTopic(String zkHosts, String topicName, int numberOfPartition, int replicationFactor, Properties properties) {
        ZkClient zkClient = null;
        try {
            zkClient = getZkClient(zkHosts);
            AdminUtils.createTopic(zkClient, topicName, numberOfPartition, replicationFactor, properties);
        } catch (Exception exception) {
            exception.printStackTrace();
        } finally {
            if (zkClient != null) {
                try {
                    zkClient.close();
                } catch (ZkInterruptedException ex) {
                    ex.printStackTrace();
                }

            }
        }
    }

    private static ZkClient getZkClient(String zkHosts) {
        ZkClient zkClient = null;
        // Zookeeper sessionTimeoutMs
        final int sessionTimeoutMs = 10000;
        // Zookeeper connectionTimeoutMs
        final int connectionTimeoutMs = 10000;
        zkClient = new ZkClient(zkHosts, sessionTimeoutMs, connectionTimeoutMs, ZKStringSerializer$.MODULE$);
        return zkClient;
    }
}

这里x.x.x.x和y.y.y.y是kafka的zk集群主机。希望这有帮助。