# Kafka to Textfiles
**URL:** https://discuss.akka.io/t/kafka-to-textfiles/896
**Category:** Akka Streams & Alpakka
**Tags:** akka, scala
**Created:** [May 2, 2018, 1:24pm UTC](https://discuss.akka.io/t/kafka-to-textfiles/896 "2018-05-02T13:24:44Z")
**Posts on this page:** 9
**Page:** 1
### Author:  [@Sharox](https://discuss.akka.io/u/Sharox)
#### Post date: [May 2, 2018, 1:24pm UTC](https://discuss.akka.io/t/kafka-to-textfiles/896/1 "2018-05-02T13:24:44Z")
Hello all,
I’m trying to work with this method: LogRotatorSink
(In the end, I would like to modify some part of code to make the method work according to my process )
You can find the implementation method here: [https://github.com/akka/alpakka/tree/master/file/src/main/scala/akka/stream/alpakka/file/scaladsl](https://github.com/akka/alpakka/tree/master/file/src/main/scala/akka/stream/alpakka/file/scaladsl)
And you can find an example here: [https://developer.lightbend.com/docs/alpakka/current/file.html](https://developer.lightbend.com/docs/alpakka/current/file.html)
I’m developing in scala and I use IntelliJ IDE
I’m copy/paste implementation method and the following message appears: “symbol Mapasync is inaccessible from this place”
In the line: “import akka.stream.impl.fusing.MapAsync.{Holder, NotYetThere}”
when I change my package name by : package akka.stream.alpakka.file.scaladsl
the message disappears ^^’
I’m newbie in scala and akka
So the problem may come from my incompetence in scala
If you don’t understand my question, I would be happy to rephrase :)
---
### Author:  [@ennru](https://discuss.akka.io/u/ennru)
#### Post date: [May 2, 2018, 1:43pm UTC](https://discuss.akka.io/t/kafka-to-textfiles/896/2 "2018-05-02T13:43:41Z")
Hi Sharox,
Thank you for reaching out with your issue, we’re happy to have you here!
So if I understand you correctly, you try to copy the implementation of `LogRotatorSink` to your code to change some details.
The implementation in Alpakka uses a bit of Akka-internal API that is not accessible by user code (as indicated by the `akka.stream.impl` package name. `MapAsync` is declared as `private[akka]`.
What is it you are trying to change in the `LogRotatorSink`?
Cheers,
Enno.
---
### Author:  [@Sharox](https://discuss.akka.io/u/Sharox)
#### Post date: [May 2, 2018, 3:07pm UTC](https://discuss.akka.io/t/kafka-to-textfiles/896/3 "2018-05-02T15:07:50Z")
Hi ennru, and thanks for your quick reply!
Ok, i understand better why this error message appears ^^
I don’t know what I need to change -\_-
But i know the process
My source is my message queue kafka
And my Path is including in the first line of my source file
(it’s why i need to change this method)
Regards,
---
### Author:  [@tg44](https://discuss.akka.io/u/tg44)
#### Post date: [May 2, 2018, 3:12pm UTC](https://discuss.akka.io/t/kafka-to-textfiles/896/4 "2018-05-02T15:12:38Z")
This connector was build with some kind of rotation in mind. So for example you want to write 100Mb to a file, and then change to an another file. Or write logs to files and each file contains one hour.
If you have a msg queue where some messages come, and every msg will go to separated file or some msges will go the same file but other goes to other I would not use this connector.
Instead I would build some kind of batching and writing thing. Something like this: [http://blog.colinbreck.com/akka-streams-a-motivating-example/](http://blog.colinbreck.com/akka-streams-a-motivating-example/) with a group by, and ofcs some kind of file sink in the end of it (which opens a file and releases it too).
---
### Author:  [@ennru](https://discuss.akka.io/u/ennru)
#### Post date: [May 3, 2018, 7:54am UTC](https://discuss.akka.io/t/kafka-to-textfiles/896/5 "2018-05-03T07:54:21Z")
You might want to look at one of the examples found on the Alpakka site:
[https://developer.lightbend.com/docs/alpakka/current/examples/jms-samples.html#example-read-text-messages-from-jms-queue-and-create-one-file-per-message](https://developer.lightbend.com/docs/alpakka/current/examples/jms-samples.html#example-read-text-messages-from-jms-queue-and-create-one-file-per-message)
It creates a one file per incoming JMS message.
---
### Author:  [@Sharox](https://discuss.akka.io/u/Sharox)
#### Post date: [May 3, 2018, 12:05pm UTC](https://discuss.akka.io/t/kafka-to-textfiles/896/6 "2018-05-03T12:05:56Z")
Hi tg44 and ennru,
Thank, for your replies
tg44, I think that’s what I want with the process that I gave you
I just have to transform it to write in a file and not in a BDD ^^
ennru, your solution is beatifull 🙂
but I need to modify my files, because 1 file contains many messages
that’s why I wanted to use LogRotatorSink method
So, I will try to take some tips from your 2 solutions
Regards,
---
### Author:  [@Sharox](https://discuss.akka.io/u/Sharox)
#### Post date: [May 3, 2018, 3:43pm UTC](https://discuss.akka.io/t/kafka-to-textfiles/896/7 "2018-05-03T15:43:27Z")
I’m sorry, but I don’t understand what i’m doing.
To start, I create my consumer :
```auto
val consumerSettings = ConsumerSettings(system, new StringDeserializer, new StringDeserializer)
.withBootstrapServers("BootstrapAdress")
.withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
.withGroupId("IdGroup")
```
Then, I create the source variable (kafka topic):
```auto
val source: Source[ConsumerMessage.CommittableMessage[String, String], Consumer.Control] = Consumer.committableSource(consumerSettings, Subscriptions.topics("MyTopic"))
```
And after, I don’t know how to do !
I need to read the first line of my source.
I don’t find on the Internet 😕
---
### Author:  [@ennru](https://discuss.akka.io/u/ennru)
#### Post date: [May 7, 2018, 7:52am UTC](https://discuss.akka.io/t/kafka-to-textfiles/896/8 "2018-05-07T07:52:40Z")
What is it you want to achieve?
You started asking about `LogRotatorSink` and are asking about the Kafka Connector now.
I assume you might want to read from Kafka and write to files?
Please take your time and play around with some of the examples I pointed at earlier:
[https://developer.lightbend.com/docs/alpakka/current/examples/index.html](https://developer.lightbend.com/docs/alpakka/current/examples/index.html)
---
### Author:  [@Sharox](https://discuss.akka.io/u/Sharox)
#### Post date: [May 7, 2018, 9:45am UTC](https://discuss.akka.io/t/kafka-to-textfiles/896/9 "2018-05-07T09:45:38Z")
Hi ennru,
Yes I started with LogRotatorSink,
But thanks to your help, I will use on other methods
(i rename the channel)
Yes, i need to read my message queue,
So, my message queue is cut into the message
The first line of my message defines the name of my csv files (or text files)
And I would like to save the message in csv files.
several messages can be part of a csv file
I looked for solutions, and I find this :
```auto
val FirstlinesStream = source.map(ByteString(_)).via(Framing.delimiter(
ByteString("\r\n"), maximumFrameLength = 100, allowTruncation = true))
.map(_.utf8String)
```
If I understand, I have the first line of my message here.
After this, I just split my line and write in my csv file