public File createSSTableFiles(String csvFile, String keyspace, String columnFamily, MutableInt lineCounter, MutableBoolean lastLineRead, BufferedReader reader) throws IOException {

        File directory = new File(keyspace+columnFamily);
        if (directory.exists()) {
            FileUtils.deleteDirectory(directory);
        }
        directory.mkdir();

        // random partitioner is created, u can give the partitioner as u want
        IPartitioner partitioner = new Murmur3Partitioner();

        List<AbstractType<?>> types = new ArrayList<AbstractType<?>>();
        types.add(AsciiType.instance);
        types.add(AsciiType.instance);
        CompositeType cmpType = CompositeType.getInstance(types);



        SSTableSimpleUnsortedWriter usersWriter = new SSTableSimpleUnsortedWriter(
                directory, partitioner, keyspace, columnFamily, cmpType, null, 64);


        String line;
        CsvEntry entry = new CsvEntry(csvFile);
        // There is no reason not to use the same timestamp for every column in that example.
        long timestamp = System.currentTimeMillis() * 1000;

        int currentCounter = 0;
        int ttl = daily_ttl;
        if(columnFamily.contains("hour"))
            ttl = hourly_ttl;


        while ((line = reader.readLine()) != null && currentCounter < LINES_IN_BATCH) {
            currentCounter++;
            if (entry.parse(line, lineCounter.intValue()+currentCounter)) {
                usersWriter.newRow(bytes(entry.partner_filter_gb1_gb2));
                usersWriter.addExpiringColumn(cmpType.builder().add(ByteBufferUtil.bytes(entry.dt_gb1_gb2)).add(ByteBufferUtil.bytes("dim_gb1")).build(), ByteBufferUtil.bytes(entry.dim_gb1), timestamp, ttl, (timestamp/1000) + ttl * 1000);
                usersWriter.addExpiringColumn(cmpType.builder().add(ByteBufferUtil.bytes(entry.dt_gb1_gb2)).add(ByteBufferUtil.bytes("dim_gb2")).build(), ByteBufferUtil.bytes(entry.dim_gb2), timestamp, ttl, (timestamp/1000) + ttl * 1000);
                usersWriter.addExpiringColumn(cmpType.builder().add(ByteBufferUtil.bytes(entry.dt_gb1_gb2)).add(ByteBufferUtil.bytes("f")).build(), ByteBufferUtil.bytes(entry.f), timestamp, ttl, (timestamp/1000) + ttl * 1000);
                for(int i=1; i < CsvEntry.max_allowed_measures;i++)
                {
                    usersWriter.addExpiringColumn(cmpType.builder().add(ByteBufferUtil.bytes(entry.dt_gb1_gb2)).add(ByteBufferUtil.bytes("m" + i)).build(), ByteBufferUtil.bytes(entry.getMeasures()[i]), timestamp, ttl, (timestamp/1000) + ttl * 1000);
                }
            }

        }
        lineCounter.add(currentCounter);
        if (reader.readLine()==null)    // last line
            lastLineRead.setValue(true);
        // Don't forget to close!
        usersWriter.close();
        return directory;
    }